diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/FakeJournalWriter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/FakeJournalWriter.cs index bb598959..c7a0eee5 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/FakeJournalWriter.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/FakeJournalWriter.cs @@ -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); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/JournalWriterTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/JournalWriterTests.cs index 35ecd14a..e3ecfba8 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/JournalWriterTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/JournalWriterTests.cs @@ -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); + } } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/SignalLogWriterTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/SignalLogWriterTests.cs index c0641c74..9a67f2b6 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/SignalLogWriterTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/SignalLogWriterTests.cs @@ -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); + } } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs index 3bc0d1f9..b02eada8 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs @@ -36,7 +36,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration /// LazyIngressSinkRelay (für die EffectRunner-↔-SignalIngress-Zirkel-Dep) /// EffectRunner (sink = Relay) /// DecisionStepProcessor (initial state) - /// SessionTraceOrdinalProvider (TODO M4.4.5.f: seed aus persisted max) + /// SessionTraceOrdinalProvider (seeded from max across SignalLog + Journal + Spool) /// SignalIngress — Relay.Target = ingress /// Scheduler.Fired → Ingress.Post-Bridge subscriben /// signalIngress.Start() — Worker-Thread läuft @@ -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(); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/IJournalWriter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/IJournalWriter.cs index 4e6ba774..b28bebc1 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/IJournalWriter.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/IJournalWriter.cs @@ -33,5 +33,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence /// Höchster bisher erfolgreich appendeter StepIndex; -1 bei leerem Journal. /// int LastStepIndex { get; } + + /// + /// Höchster in irgendeiner persisted Transition vorkommender SessionTraceOrdinal; + /// -1 bei leerem Journal. Analog ISignalLogWriter.LastTraceOrdinal — wird vom + /// Orchestrator für das Seeding von SessionTraceOrdinalProvider auf Recovery + /// verwendet. + /// + long LastTraceOrdinal { get; } } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/ISignalLogWriter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/ISignalLogWriter.cs index 62198e25..78f2ee92 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/ISignalLogWriter.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/ISignalLogWriter.cs @@ -37,5 +37,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence /// Höchster bisher erfolgreich appendete SessionSignalOrdinal; -1 bei leerem Log. /// long LastOrdinal { get; } + + /// + /// Höchster in irgendeinem persisted Signal vorkommender SessionTraceOrdinal; -1 bei + /// leerem Log. Wird von EnrollmentOrchestrator zum Seed des + /// SessionTraceOrdinalProvider auf Recovery benutzt — zusammen mit + /// IJournalWriter.LastTraceOrdinal und TelemetrySpool.LastAssignedItemId. + /// + /// Unterschied zu : SessionSignalOrdinal ist der + /// Signal-spezifische Monotone-Counter, SessionTraceOrdinal der session-weite + /// Korrelations-Counter (Event + Signal + Transition teilen sich eine Nummerierung). + /// + /// + long LastTraceOrdinal { get; } } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/JournalWriter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/JournalWriter.cs index 32b6e616..52829cee 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/JournalWriter.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/JournalWriter.cs @@ -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() + /// + /// Single-pass scan that populates both _lastStepIndex and + /// _lastTraceOrdinal from the persisted JSONL. Stops at the first unparsable + /// line (recovery §2.7). + /// + 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; } } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SignalLogWriter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SignalLogWriter.cs index 28f56480..bbff7740 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SignalLogWriter.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SignalLogWriter.cs @@ -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() + /// + /// Single-pass scan that populates both _lastOrdinal and _lastTraceOrdinal + /// from the persisted JSONL. Stops at the first unparsable line (recovery §2.7). + /// + 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; } } }