From b6ea13f122f5818a823a7f0cfee7861d0d7b9717 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Tue, 21 Apr 2026 14:31:28 +0200 Subject: [PATCH] fix(agent): upload V2 Signals + DecisionTransitions to backend (Codex Finding 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The V2 agent was never projecting DecisionSignals or DecisionTransitions onto the telemetry transport. SignalIngress.Append wrote the signal to the local SignalLog and handed it to the reducer; DecisionStepProcessor journaled each transition locally. Neither called into ITelemetryTransport, so the entire M5 backend ingest pipeline for these two telemetry kinds (Signals/DecisionTransitions primary tables, Inspector reads, ReducerVerifier, the M5.d index fan-out + timer) had zero agent traffic. Adds the missing emitters — analogous to TelemetryEventEmitter — and wires them through EnrollmentOrchestrator: - TelemetrySignalEmitter (V2.Core/Telemetry/Signals/): serializes a DecisionSignal with Newtonsoft + StringEnumConverter and builds a TelemetryItemDraft with Kind=Signal, PK={tenant}_{session}, RK=SessionSignalOrdinal.ToString("D19") — matches TableSignalRepository.BuildRowKey for idempotent UpsertReplace. - TelemetryTransitionEmitter (V2.Core/Telemetry/Transitions/): same contract for DecisionTransition; RK=StepIndex.ToString("D10"). Wiring: - SignalIngress accepts an optional TelemetrySignalEmitter. After the local SignalLog.Append succeeds — before the reducer runs — the ingress enqueues the signal on the transport. Enqueue failures are swallowed: the local SignalLog is authoritative per §2.7c / L.1, and the reducer must still see the signal even if the upload path is broken. - DecisionStepProcessor accepts an optional TelemetryTransitionEmitter. After the Journal.Append try-block succeeds (and before the effect runner), it enqueues the transition. Same swallow semantics. Journal failures still rethrow as before, so a failed-journal step can never leak a ghost row to the backend. - EnrollmentOrchestrator constructs both emitters alongside the existing TelemetryEventEmitter (step 4 of the startup sequence) and passes them into steps 8 + 10. Both emitter constructor params are optional to keep pre-M5 test constructors compilable without touch-up. Tests (+26 across two assemblies): V2.Core.Tests (+21): - TelemetrySignalEmitterTests (6): correct keys + kind, StringEnumConverter for DecisionSignalKind, PascalCase field names the backend parser reads, D19 padding survives long.MaxValue, null-rejection, ctor argument validation. - TelemetryTransitionEmitterTests (7): same surface + dead-end case preserves DeadEndReason + Taken=false, nested ClassifierVerdict emits ClassifierId/Level that the backend parser can read. - SignalIngressUploadPathTests (4): emit happens per signal, monotonic RowKey padding, null emitter preserves pre-M5 behaviour, transport throw is swallowed (reducer keeps running). - DecisionStepProcessorUploadPathTests (4): emit happens per successful journal, null emitter preserves pre-M5 behaviour, journal failure prevents emit (no ghost rows), transport throw is swallowed (effects still run + CurrentState advances). Functions.Tests (+5) — the Codex-requested cross-layer gate: - TelemetryWireRoundTripTests: a DecisionSignal / DecisionTransition serialized with the exact JsonSerializerSettings the agent emitter uses (StringEnumConverter + NullValueHandling.Ignore) is parsed by TelemetryPayloadParser.ParseSignal / ParseTransition back into a SignalRecord / DecisionTransitionRecord with every field preserved. Includes explicit-zero edge cases (ordinal=0, stepIndex=0) so the Finding-3 required-field guards don't trip on legitimate agent traffic. The backend project can't reference the net48 agent assembly directly, so the test replicates the wire contract via shared DecisionCore types + the agreed serializer settings — any drift between agent and backend breaks this test immediately. Suites: V2.Core 406 green (prev 385 + 21 new). Functions 708 green (prev 703 + 5 new). Full solution build green. Does not address Codex Finding 2 (AdminAction=Succeeded mis-mapped to Failed locally) — lands in the next commit on this branch. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../DecisionStepProcessorUploadPathTests.cs | 147 +++++++++++++ .../SignalIngressUploadPathTests.cs | 157 +++++++++++++ .../Signals/TelemetrySignalEmitterTests.cs | 113 ++++++++++ .../TelemetryTransitionEmitterTests.cs | 168 ++++++++++++++ .../Orchestration/DecisionStepProcessor.cs | 15 +- .../Orchestration/EnrollmentOrchestrator.cs | 10 +- .../Orchestration/SignalIngress.cs | 14 ++ .../Signals/TelemetrySignalEmitter.cs | 71 ++++++ .../Transitions/TelemetryTransitionEmitter.cs | 72 ++++++ .../TelemetryWireRoundTripTests.cs | 206 ++++++++++++++++++ 10 files changed, 971 insertions(+), 2 deletions(-) create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DecisionStepProcessorUploadPathTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressUploadPathTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Signals/TelemetrySignalEmitterTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Transitions/TelemetryTransitionEmitterTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Signals/TelemetrySignalEmitter.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Transitions/TelemetryTransitionEmitter.cs create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/TelemetryWireRoundTripTests.cs diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DecisionStepProcessorUploadPathTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DecisionStepProcessorUploadPathTests.cs new file mode 100644 index 00000000..8d2a41da --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DecisionStepProcessorUploadPathTests.cs @@ -0,0 +1,147 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Orchestration; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Transitions; +using AutopilotMonitor.Agent.V2.Core.Tests.Harness; +using AutopilotMonitor.Agent.V2.Core.Tests.Telemetry.Events; +using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Signals; +using AutopilotMonitor.DecisionCore.State; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration +{ + /// + /// Codex Finding 1 regression gate for : verifies that + /// every successfully-journaled is also projected onto + /// the telemetry transport for backend upload, and that a throwing emitter does NOT + /// abort the step. + /// + public sealed class DecisionStepProcessorUploadPathTests + { + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + + private static DateTime At => new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc); + + private sealed class Rig : IDisposable + { + public TempDirectory Tmp { get; } = new TempDirectory(); + public AgentLogger Logger { get; } + public FakeJournalWriter Journal { get; } = new FakeJournalWriter(); + public FakeEffectRunner Effects { get; } = new FakeEffectRunner(); + public FakeSnapshotPersistence Snapshot { get; } = new FakeSnapshotPersistence(); + public FakeQuarantineSink Quarantine { get; } = new FakeQuarantineSink(); + public DecisionEngine Engine { get; } = new DecisionEngine(); + public DecisionState InitialState { get; } = DecisionState.CreateInitial(SessionId, TenantId); + public FakeTelemetryTransport Transport { get; } = new FakeTelemetryTransport(); + + public Rig() { Logger = new AgentLogger(Tmp.Path, AgentLogLevel.Info); } + + public DecisionStepProcessor Build(TelemetryTransitionEmitter? emitter) + => new DecisionStepProcessor( + initialState: InitialState, + journal: Journal, + effectRunner: Effects, + snapshot: Snapshot, + quarantineSink: Quarantine, + logger: Logger, + transitionEmitter: emitter); + + public (DecisionStep step, DecisionSignal signal) ReduceSessionStarted(long ordinal = 0) + { + var signal = TestSignals.Raw(ordinal, DecisionSignalKind.SessionStarted, At); + var step = Engine.Reduce(InitialState, signal); + return (step, signal); + } + + public void Dispose() => Tmp.Dispose(); + } + + [Fact] + public void ApplyStep_emits_transition_to_transport_after_Journal_append() + { + using var rig = new Rig(); + var emitter = new TelemetryTransitionEmitter(rig.Transport, SessionId, TenantId); + var processor = rig.Build(emitter); + + var (step, signal) = rig.ReduceSessionStarted(); + processor.ApplyStep(step, signal); + + Assert.Single(rig.Journal.Appended); + Assert.Equal(1, rig.Transport.EnqueueCount); + Assert.Equal(TelemetryItemKind.DecisionTransition, rig.Transport.Enqueued[0].Kind); + Assert.Equal($"{TenantId}_{SessionId}", rig.Transport.Enqueued[0].PartitionKey); + } + + [Fact] + public void ApplyStep_without_emitter_keeps_pre_M5_behaviour_unchanged() + { + using var rig = new Rig(); + var processor = rig.Build(emitter: null); + + var (step, signal) = rig.ReduceSessionStarted(); + processor.ApplyStep(step, signal); + + Assert.Single(rig.Journal.Appended); + Assert.Equal(0, rig.Transport.EnqueueCount); + } + + [Fact] + public void Journal_failure_prevents_emit_so_no_ghost_rows_leak_to_backend() + { + // Journal is authoritative. If journal.Append throws, the transition NEVER happened + // locally — and must NEVER be uploaded. Otherwise the backend would see phantom + // transitions that can't be reconciled. + using var rig = new Rig(); + rig.Journal.ScriptThrow(new InvalidOperationException("disk full (test)")); + var emitter = new TelemetryTransitionEmitter(rig.Transport, SessionId, TenantId); + var processor = rig.Build(emitter); + + var (step, signal) = rig.ReduceSessionStarted(); + + Assert.Throws(() => processor.ApplyStep(step, signal)); + Assert.Equal(0, rig.Transport.EnqueueCount); + } + + [Fact] + public void Transport_throw_is_swallowed_and_effects_still_run() + { + // Transport failure must NOT break the local post-journal pipeline: snapshot + + // effects + CurrentState update all happen regardless. + using var rig = new Rig(); + var throwingTransport = new ThrowingTelemetryTransport(); + var emitter = new TelemetryTransitionEmitter(throwingTransport, SessionId, TenantId); + var processor = rig.Build(emitter); + + var (step, signal) = rig.ReduceSessionStarted(); + processor.ApplyStep(step, signal); + + Assert.Equal(1, throwingTransport.EnqueueAttempts); + Assert.Single(rig.Journal.Appended); + Assert.Equal(1, rig.Effects.CallCount); + Assert.Equal(step.NewState, processor.CurrentState); + } + + private sealed class ThrowingTelemetryTransport : ITelemetryTransport + { + public int EnqueueAttempts; + + public long LastUploadedItemId => -1; + + public TelemetryItem Enqueue(TelemetryItemDraft draft) + { + Interlocked.Increment(ref EnqueueAttempts); + throw new InvalidOperationException("spool overflow (test)"); + } + + public Task DrainAllAsync(CancellationToken cancellationToken = default) + => Task.FromResult(DrainResult.Empty()); + + public void Dispose() { } + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressUploadPathTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressUploadPathTests.cs new file mode 100644 index 00000000..5f0ed1e5 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressUploadPathTests.cs @@ -0,0 +1,157 @@ +using System; +using System.Threading; +using AutopilotMonitor.Agent.V2.Core.Orchestration; +using AutopilotMonitor.Agent.V2.Core.Persistence; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Signals; +using AutopilotMonitor.Agent.V2.Core.Tests.Harness; +using AutopilotMonitor.Agent.V2.Core.Tests.Telemetry.Events; +using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Signals; +using Xunit; + +#pragma warning disable xUnit1031 + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration +{ + /// + /// Codex Finding 1 regression gate: without the emitter hook in + /// , no signal ever reached the telemetry transport and the + /// backend /api/sessions/{id}/signals endpoint saw zero traffic. These tests + /// verify that every accepted signal also lands on the transport, and that a throwing + /// transport does NOT take down the ingress worker. + /// + [Collection("SerialThreading")] + public sealed class SignalIngressUploadPathTests + { + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + + private static DateTime At => new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc); + + private static Evidence RawEvidence(string id = "raw-1") => + new Evidence(EvidenceKind.Raw, id, $"evidence-{id}"); + + private sealed class Rig : IDisposable + { + public TempDirectory Tmp { get; } = new TempDirectory(); + public SignalLogWriter SignalLog { get; } + public SessionTraceOrdinalProvider TraceCounter { get; } = new SessionTraceOrdinalProvider(); + public FakeDecisionStepProcessor Processor { get; } = new FakeDecisionStepProcessor(); + public VirtualClock Clock { get; } = new VirtualClock(At); + public DecisionEngine Engine { get; } = new DecisionEngine(); + public FakeTelemetryTransport Transport { get; } = new FakeTelemetryTransport(); + + public Rig() { SignalLog = new SignalLogWriter(Tmp.File("signal-log.jsonl")); } + + public SignalIngress Build(TelemetrySignalEmitter? emitter) + => new SignalIngress( + engine: Engine, + signalLog: SignalLog, + traceCounter: TraceCounter, + processor: Processor, + clock: Clock, + backPressureObserver: null, + signalEmitter: emitter); + + public void Dispose() => Tmp.Dispose(); + } + + private static bool WaitFor(Func cond, int timeoutMs = 5000) + => SpinWait.SpinUntil(cond, timeoutMs); + + [Fact] + public void Post_emits_to_transport_after_SignalLog_append_succeeds() + { + using var rig = new Rig(); + var emitter = new TelemetrySignalEmitter(rig.Transport, SessionId, TenantId); + var ing = rig.Build(emitter); + ing.Start(); + + ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence("e-1")); + + Assert.True(WaitFor(() => rig.Transport.EnqueueCount == 1)); + Assert.Equal(TelemetryItemKind.Signal, rig.Transport.Enqueued[0].Kind); + Assert.Equal($"{TenantId}_{SessionId}", rig.Transport.Enqueued[0].PartitionKey); + + ing.Stop(); + } + + [Fact] + public void Post_emits_one_transport_item_per_signal() + { + using var rig = new Rig(); + var emitter = new TelemetrySignalEmitter(rig.Transport, SessionId, TenantId); + var ing = rig.Build(emitter); + ing.Start(); + + for (int i = 0; i < 5; i++) + { + ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence($"e-{i}")); + } + + Assert.True(WaitFor(() => rig.Transport.EnqueueCount == 5)); + // RowKeys are D19(ordinal) — SignalIngress assigns ordinals 0..4 against an empty log. + for (int i = 0; i < 5; i++) + { + Assert.Equal(((long)i).ToString("D19"), rig.Transport.Enqueued[i].RowKey); + } + + ing.Stop(); + } + + [Fact] + public void Post_without_emitter_keeps_pre_M5_behaviour_unchanged() + { + // emitter=null (the default) means the agent runs with SignalLog-only persistence. + // Regression guard so the optional-param wiring never becomes required. + using var rig = new Rig(); + var ing = rig.Build(emitter: null); + ing.Start(); + + ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence()); + + Assert.True(WaitFor(() => rig.Processor.ApplyCallCount == 1)); + Assert.Equal(0, rig.Transport.EnqueueCount); + + ing.Stop(); + } + + [Fact] + public void Transport_throw_is_swallowed_and_ingress_keeps_reducing() + { + using var rig = new Rig(); + var throwingTransport = new ThrowingTelemetryTransport(); + var emitter = new TelemetrySignalEmitter(throwingTransport, SessionId, TenantId); + var ing = rig.Build(emitter); + ing.Start(); + + // Even with the transport failing on every Enqueue, the reducer must still see the + // signal — local SignalLog is authoritative per §2.7c. + ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence()); + + Assert.True(WaitFor(() => rig.Processor.ApplyCallCount == 1)); + Assert.Equal(1, throwingTransport.EnqueueAttempts); + + ing.Stop(); + } + + private sealed class ThrowingTelemetryTransport : ITelemetryTransport + { + public int EnqueueAttempts; + + public long LastUploadedItemId => -1; + + public TelemetryItem Enqueue(TelemetryItemDraft draft) + { + Interlocked.Increment(ref EnqueueAttempts); + throw new InvalidOperationException("spool overflow (test)"); + } + + public System.Threading.Tasks.Task DrainAllAsync(CancellationToken cancellationToken = default) + => System.Threading.Tasks.Task.FromResult(DrainResult.Empty()); + + public void Dispose() { } + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Signals/TelemetrySignalEmitterTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Signals/TelemetrySignalEmitterTests.cs new file mode 100644 index 00000000..52b6194d --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Signals/TelemetrySignalEmitterTests.cs @@ -0,0 +1,113 @@ +using System; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Signals; +using AutopilotMonitor.Agent.V2.Core.Tests.Telemetry.Events; +using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry; +using AutopilotMonitor.DecisionCore.Signals; +using Newtonsoft.Json.Linq; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Telemetry.Signals +{ + /// + /// Pure emitter tests for (Codex Finding 1 fix). + /// Verifies the agent-side wire format and transport projection; the cross-layer + /// round-trip with the backend parser lives in TelemetrySignalRoundTripTests. + /// + public sealed class TelemetrySignalEmitterTests + { + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + + private static readonly DateTime At = new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc); + + private static DecisionSignal Signal(long ordinal, DecisionSignalKind kind = DecisionSignalKind.EspPhaseChanged) + => new DecisionSignal( + sessionSignalOrdinal: ordinal, + sessionTraceOrdinal: ordinal + 100, + kind: kind, + kindSchemaVersion: 1, + occurredAtUtc: At, + sourceOrigin: "EspAndHelloTrackerAdapter", + evidence: new Evidence(EvidenceKind.Raw, $"raw-{ordinal}", $"evidence-{ordinal}")); + + [Fact] + public void Emit_routes_signal_through_transport_with_correct_keys() + { + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetrySignalEmitter(transport, SessionId, TenantId); + + emitter.Emit(Signal(42)); + + Assert.Equal(1, transport.EnqueueCount); + var item = transport.Enqueued[0]; + Assert.Equal(TelemetryItemKind.Signal, item.Kind); + Assert.Equal($"{TenantId}_{SessionId}", item.PartitionKey); + Assert.Equal("0000000000000000042", item.RowKey); // D19 padding — matches TableSignalRepository.BuildRowKey + Assert.False(item.RequiresImmediateFlush); + } + + [Fact] + public void Emit_serializes_Kind_as_string_enum_name_for_backend_parser_compatibility() + { + // Backend TelemetryPayloadParser.ParseSignal reads `(string?)root["Kind"]`. Default + // Newtonsoft serializes enums as ints, which would silently fail the cast. The emitter + // must configure StringEnumConverter. + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetrySignalEmitter(transport, SessionId, TenantId); + + emitter.Emit(Signal(1, DecisionSignalKind.EspTerminalFailure)); + + var parsed = JObject.Parse(transport.Enqueued[0].PayloadJson); + Assert.Equal("EspTerminalFailure", (string?)parsed["Kind"]); + } + + [Fact] + public void Emit_emits_PascalCase_fields_the_backend_parser_reads() + { + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetrySignalEmitter(transport, SessionId, TenantId); + + emitter.Emit(Signal(7)); + + var parsed = JObject.Parse(transport.Enqueued[0].PayloadJson); + Assert.Equal(7L, (long?)parsed["SessionSignalOrdinal"]); + Assert.Equal(107L, (long?)parsed["SessionTraceOrdinal"]); + Assert.Equal("EspPhaseChanged", (string?)parsed["Kind"]); + Assert.Equal(1, (int?)parsed["KindSchemaVersion"]); + Assert.Equal("EspAndHelloTrackerAdapter", (string?)parsed["SourceOrigin"]); + Assert.NotNull(parsed["OccurredAtUtc"]); + } + + [Fact] + public void Emit_RowKey_padding_survives_long_MaxValue() + { + // 19 digits covers long.MaxValue. Guard the invariant — lex ordering matches numeric + // ordinal order only with full padding. + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetrySignalEmitter(transport, SessionId, TenantId); + + emitter.Emit(Signal(long.MaxValue)); + + Assert.Equal("9223372036854775807", transport.Enqueued[0].RowKey); + } + + [Fact] + public void Emit_rejects_null_signal() + { + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetrySignalEmitter(transport, SessionId, TenantId); + + Assert.Throws(() => emitter.Emit(null!)); + Assert.Equal(0, transport.EnqueueCount); + } + + [Fact] + public void Ctor_rejects_null_or_empty_identifiers() + { + var transport = new FakeTelemetryTransport(); + Assert.Throws(() => new TelemetrySignalEmitter(transport, "", TenantId)); + Assert.Throws(() => new TelemetrySignalEmitter(transport, SessionId, "")); + Assert.Throws(() => new TelemetrySignalEmitter(null!, SessionId, TenantId)); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Transitions/TelemetryTransitionEmitterTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Transitions/TelemetryTransitionEmitterTests.cs new file mode 100644 index 00000000..f6a9499f --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Transitions/TelemetryTransitionEmitterTests.cs @@ -0,0 +1,168 @@ +using System; +using System.Collections.Generic; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Transitions; +using AutopilotMonitor.Agent.V2.Core.Tests.Telemetry.Events; +using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry; +using AutopilotMonitor.DecisionCore.Classifiers; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.State; +using Newtonsoft.Json.Linq; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Telemetry.Transitions +{ + /// + /// Pure emitter tests for (Codex Finding 1 fix). + /// Cross-layer round-trip with the backend parser lives in + /// TelemetryTransitionRoundTripTests. + /// + public sealed class TelemetryTransitionEmitterTests + { + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + + private static readonly DateTime At = new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc); + + private static DecisionTransition Taken(int stepIndex, SessionStage to = SessionStage.EspDeviceSetup) + => new DecisionTransition( + stepIndex: stepIndex, + sessionTraceOrdinal: stepIndex + 500, + signalOrdinalRef: stepIndex + 1, + occurredAtUtc: At, + trigger: "EspExiting", + fromStage: SessionStage.AwaitingEspPhaseChange, + toStage: to, + taken: true, + deadEndReason: null, + reducerVersion: "1.0.0"); + + [Fact] + public void Emit_routes_transition_through_transport_with_correct_keys() + { + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetryTransitionEmitter(transport, SessionId, TenantId); + + emitter.Emit(Taken(7)); + + Assert.Equal(1, transport.EnqueueCount); + var item = transport.Enqueued[0]; + Assert.Equal(TelemetryItemKind.DecisionTransition, item.Kind); + Assert.Equal($"{TenantId}_{SessionId}", item.PartitionKey); + Assert.Equal("0000000007", item.RowKey); // D10 padding — matches TableDecisionTransitionRepository.BuildRowKey + Assert.False(item.RequiresImmediateFlush); + } + + [Fact] + public void Emit_serializes_FromStage_and_ToStage_as_string_enum_names() + { + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetryTransitionEmitter(transport, SessionId, TenantId); + + emitter.Emit(Taken(1, to: SessionStage.WhiteGloveSealed)); + + var parsed = JObject.Parse(transport.Enqueued[0].PayloadJson); + Assert.Equal("AwaitingEspPhaseChange", (string?)parsed["FromStage"]); + Assert.Equal("WhiteGloveSealed", (string?)parsed["ToStage"]); + } + + [Fact] + public void Emit_emits_PascalCase_fields_the_backend_parser_reads() + { + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetryTransitionEmitter(transport, SessionId, TenantId); + + emitter.Emit(Taken(3)); + + var parsed = JObject.Parse(transport.Enqueued[0].PayloadJson); + Assert.Equal(3, (int?)parsed["StepIndex"]); + Assert.Equal(503L, (long?)parsed["SessionTraceOrdinal"]); + Assert.Equal(4L, (long?)parsed["SignalOrdinalRef"]); + Assert.Equal("EspExiting", (string?)parsed["Trigger"]); + Assert.Equal(true, (bool?)parsed["Taken"]); + Assert.Equal("1.0.0", (string?)parsed["ReducerVersion"]); + Assert.NotNull(parsed["OccurredAtUtc"]); + } + + [Fact] + public void Emit_dead_end_preserves_DeadEndReason_and_Taken_false() + { + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetryTransitionEmitter(transport, SessionId, TenantId); + + var blocked = new DecisionTransition( + stepIndex: 5, + sessionTraceOrdinal: 505, + signalOrdinalRef: 6, + occurredAtUtc: At, + trigger: "EspExiting", + fromStage: SessionStage.EspDeviceSetup, + toStage: SessionStage.EspDeviceSetup, + taken: false, + deadEndReason: "hybrid_reboot_gate_blocking", + reducerVersion: "1.0.0"); + + emitter.Emit(blocked); + + var parsed = JObject.Parse(transport.Enqueued[0].PayloadJson); + Assert.Equal(false, (bool?)parsed["Taken"]); + Assert.Equal("hybrid_reboot_gate_blocking", (string?)parsed["DeadEndReason"]); + } + + [Fact] + public void Emit_with_classifier_verdict_emits_nested_object_parser_can_read() + { + // Backend ParseTransition reads ClassifierVerdict.ClassifierId and ClassifierVerdict.Level + // (also accepts "HypothesisLevel" as fallback). HypothesisLevel enum must serialize as + // a string — StringEnumConverter handles it. + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetryTransitionEmitter(transport, SessionId, TenantId); + + var verdict = new ClassifierVerdict( + classifierId: "whiteglove-sealing", + level: HypothesisLevel.Strong, + score: 80, + contributingFactors: new List { "PatternMatched" }, + reason: "test", + inputHash: "abc123"); + + var withVerdict = new DecisionTransition( + stepIndex: 9, + sessionTraceOrdinal: 509, + signalOrdinalRef: 10, + occurredAtUtc: At, + trigger: "EspExiting", + fromStage: SessionStage.EspDeviceSetup, + toStage: SessionStage.WhiteGloveSealed, + taken: true, + deadEndReason: null, + reducerVersion: "1.0.0", + classifierVerdict: verdict); + + emitter.Emit(withVerdict); + + var parsed = JObject.Parse(transport.Enqueued[0].PayloadJson); + var verdictJson = (JObject)parsed["ClassifierVerdict"]!; + Assert.Equal("whiteglove-sealing", (string?)verdictJson["ClassifierId"]); + Assert.Equal("Strong", (string?)verdictJson["Level"]); + } + + [Fact] + public void Emit_rejects_null_transition() + { + var transport = new FakeTelemetryTransport(); + var emitter = new TelemetryTransitionEmitter(transport, SessionId, TenantId); + + Assert.Throws(() => emitter.Emit(null!)); + Assert.Equal(0, transport.EnqueueCount); + } + + [Fact] + public void Ctor_rejects_null_or_empty_identifiers() + { + var transport = new FakeTelemetryTransport(); + Assert.Throws(() => new TelemetryTransitionEmitter(transport, "", TenantId)); + Assert.Throws(() => new TelemetryTransitionEmitter(transport, SessionId, "")); + Assert.Throws(() => new TelemetryTransitionEmitter(null!, SessionId, TenantId)); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs index cbc49356..47ec88ea 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs @@ -3,6 +3,7 @@ using System; using System.Threading; using AutopilotMonitor.Agent.V2.Core.Logging; using AutopilotMonitor.Agent.V2.Core.Persistence; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Transitions; using AutopilotMonitor.DecisionCore.Engine; using AutopilotMonitor.DecisionCore.Signals; using AutopilotMonitor.DecisionCore.State; @@ -51,6 +52,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration private readonly AgentLogger _logger; private readonly int _quarantineThreshold; private readonly Action? _onTerminalStageReached; + private readonly TelemetryTransitionEmitter? _transitionEmitter; private DecisionState _currentState; private int _consecutiveJournalFailures; @@ -65,7 +67,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration IQuarantineSink quarantineSink, AgentLogger logger, int quarantineThreshold = DefaultQuarantineThreshold, - Action? onTerminalStageReached = null) + Action? onTerminalStageReached = null, + TelemetryTransitionEmitter? transitionEmitter = null) { if (quarantineThreshold <= 0) { @@ -82,6 +85,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _quarantineThreshold = quarantineThreshold; _onTerminalStageReached = onTerminalStageReached; + _transitionEmitter = transitionEmitter; // If recovery loaded a state that already sits on a terminal stage (e.g. a crash // after a success-path step but before Stop()), treat it as already-notified so we @@ -123,6 +127,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration throw; } + // 1a) Project the transition onto the telemetry transport for backend upload. Journal + // is authoritative (§2.7c / L.1); a transport enqueue failure here must NOT abort the + // step — the local journal already committed and effects below must still run. + if (_transitionEmitter != null) + { + try { _transitionEmitter.Emit(step.Transition); } + catch { /* best-effort upload; local state already consistent */ } + } + // 2) EffectRunner — Async-Methode vom Ingress-Worker-Thread sync ausführen. // Kein SynchronizationContext auf einem reinen Background-Thread, deshalb // GetAwaiter().GetResult() hier deadlock-frei (L.1 Ingress-Worker-Thread). diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs index 1891b38f..3bc0d1f9 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs @@ -7,6 +7,8 @@ using System.Threading.Tasks; using AutopilotMonitor.Agent.V2.Core.Logging; using AutopilotMonitor.Agent.V2.Core.Persistence; using AutopilotMonitor.Agent.V2.Core.Telemetry.Events; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Signals; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Transitions; using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry; using AutopilotMonitor.DecisionCore.Classifiers; using AutopilotMonitor.DecisionCore.Engine; @@ -92,6 +94,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration private TelemetrySpool? _spool; private TelemetryUploadOrchestrator? _transport; private TelemetryEventEmitter? _eventEmitter; + private TelemetrySignalEmitter? _signalEmitter; + private TelemetryTransitionEmitter? _transitionEmitter; private EventTimelineEmitter? _timelineEmitter; private BackPressureEventObserver? _backPressureObserver; private DeadlineScheduler? _scheduler; @@ -309,6 +313,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration // 4) Event-Emitter-Kette. _eventSequenceCounter = new EventSequenceCounter(_eventSequencePersistence); _eventEmitter = new TelemetryEventEmitter(_transport, _eventSequenceCounter, _sessionId, _tenantId); + _signalEmitter = new TelemetrySignalEmitter(_transport, _sessionId, _tenantId); + _transitionEmitter = new TelemetryTransitionEmitter(_transport, _sessionId, _tenantId); _timelineEmitter = new EventTimelineEmitter(_eventEmitter); _backPressureObserver = new BackPressureEventObserver(_eventEmitter, _clock); @@ -340,7 +346,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration quarantineSink: this, logger: _logger, quarantineThreshold: _quarantineThreshold, - onTerminalStageReached: OnDecisionTerminalStage); + onTerminalStageReached: OnDecisionTerminalStage, + transitionEmitter: _transitionEmitter); // 9) Trace-Ordinal. TODO(M4.4.5.f): seed aus max(SignalLog.LastOrdinal, // Journal.LastStepIndex, Spool.LastAssignedItemId via SessionTraceOrdinal). @@ -355,6 +362,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration processor: _processor, clock: _clock, backPressureObserver: _backPressureObserver, + signalEmitter: _signalEmitter, channelCapacity: _channelCapacity); // 11) Relay auf den echten Ingress umbiegen. diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs index e916687d..7c8a64d0 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs @@ -4,6 +4,7 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Threading; using AutopilotMonitor.Agent.V2.Core.Persistence; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Signals; using AutopilotMonitor.DecisionCore.Engine; using AutopilotMonitor.DecisionCore.Signals; @@ -43,6 +44,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration private readonly ISessionTraceOrdinalProvider _traceCounter; private readonly IDecisionStepProcessor _processor; private readonly IBackPressureObserver? _observer; + private readonly TelemetrySignalEmitter? _signalEmitter; private readonly IClock _clock; private readonly TimeSpan _backPressureThrottle; private readonly int _channelCapacity; @@ -65,6 +67,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration IDecisionStepProcessor processor, IClock clock, IBackPressureObserver? backPressureObserver = null, + TelemetrySignalEmitter? signalEmitter = null, int channelCapacity = DefaultChannelCapacity, TimeSpan? backPressureThrottle = null) { @@ -79,6 +82,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration _processor = processor ?? throw new ArgumentNullException(nameof(processor)); _clock = clock ?? throw new ArgumentNullException(nameof(clock)); _observer = backPressureObserver; + _signalEmitter = signalEmitter; _channelCapacity = channelCapacity; _backPressureThrottle = backPressureThrottle ?? DefaultBackPressureThrottle; @@ -291,6 +295,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration _nextSignalOrdinal = signalOrdinal + 1; + // Project the signal onto the telemetry transport for backend upload. Local + // SignalLog is authoritative (§2.7c / L.1); a transport enqueue failure must NOT + // break the ingress worker. Spool overflow / serialization faults get swallowed + // here so the reducer path below still runs on the just-committed local signal. + if (_signalEmitter != null) + { + try { _signalEmitter.Emit(signal); } + catch { /* best-effort upload; local state already consistent */ } + } + var state = _processor.CurrentState; // IDecisionEngine.Reduce ist fail-safe (DecisionEngine.cs:39-47): wirft nicht. var step = _engine.Reduce(state, signal); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Signals/TelemetrySignalEmitter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Signals/TelemetrySignalEmitter.cs new file mode 100644 index 00000000..c9739692 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Signals/TelemetrySignalEmitter.cs @@ -0,0 +1,71 @@ +#nullable enable +using System; +using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry; +using AutopilotMonitor.DecisionCore.Signals; +using Newtonsoft.Json; +using Newtonsoft.Json.Converters; + +namespace AutopilotMonitor.Agent.V2.Core.Telemetry.Signals +{ + /// + /// Emitter that projects s onto the generic + /// for upload to the backend /api/agent/telemetry + /// endpoint (Plan §2.7a, M5.a Signals table). Sibling of + /// . + /// + /// Called by after SignalLog.Append has + /// committed locally — the local persistence is authoritative (§2.7c / L.1), the upload + /// is a best-effort projection. Enqueue failures are the caller's to swallow. + /// + /// + /// Wire format: Newtonsoft PascalCase with for + /// . Matches the backend-side shape + /// TelemetryPayloadParser.ParseSignal expects (SessionSignalOrdinal, + /// Kind, KindSchemaVersion, OccurredAtUtc, SourceOrigin, + /// SessionTraceOrdinal). + /// + /// + public sealed class TelemetrySignalEmitter + { + private static readonly JsonSerializerSettings WireFormatSettings = new JsonSerializerSettings + { + Converters = { new StringEnumConverter() }, + NullValueHandling = NullValueHandling.Ignore, + }; + + private readonly ITelemetryTransport _transport; + private readonly string _partitionKey; + + public TelemetrySignalEmitter(ITelemetryTransport transport, string sessionId, string tenantId) + { + if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("SessionId is mandatory.", nameof(sessionId)); + if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId is mandatory.", nameof(tenantId)); + _transport = transport ?? throw new ArgumentNullException(nameof(transport)); + _partitionKey = $"{tenantId}_{sessionId}"; + } + + /// + /// Serialises the signal and enqueues it as a + /// draft. RowKey is the 19-digit zero-padded + /// — matches TableSignalRepository.BuildRowKey on the backend so Upsert collapses + /// duplicates. + /// + public TelemetryItem Emit(DecisionSignal signal) + { + if (signal == null) throw new ArgumentNullException(nameof(signal)); + + var payloadJson = JsonConvert.SerializeObject(signal, Formatting.None, WireFormatSettings); + var rowKey = signal.SessionSignalOrdinal.ToString("D19"); + + var draft = new TelemetryItemDraft( + kind: TelemetryItemKind.Signal, + partitionKey: _partitionKey, + rowKey: rowKey, + payloadJson: payloadJson, + isSessionScoped: true, + requiresImmediateFlush: false); + + return _transport.Enqueue(draft); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Transitions/TelemetryTransitionEmitter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Transitions/TelemetryTransitionEmitter.cs new file mode 100644 index 00000000..69ee6fa7 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Transitions/TelemetryTransitionEmitter.cs @@ -0,0 +1,72 @@ +#nullable enable +using System; +using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry; +using AutopilotMonitor.DecisionCore.Engine; +using Newtonsoft.Json; +using Newtonsoft.Json.Converters; + +namespace AutopilotMonitor.Agent.V2.Core.Telemetry.Transitions +{ + /// + /// Emitter that projects s onto the generic + /// for upload to the backend /api/agent/telemetry + /// endpoint (Plan §2.7a, M5.a DecisionTransitions table). Sibling of + /// / . + /// + /// Called by after + /// Journal.Append(transition) has committed locally — Journal is the authoritative + /// store (§2.7c / L.1), upload is a best-effort projection. + /// + /// + /// Wire format: Newtonsoft PascalCase with for + /// SessionStage, HypothesisLevel, etc. Matches the backend shape + /// TelemetryPayloadParser.ParseTransition expects (StepIndex, + /// SessionTraceOrdinal, SignalOrdinalRef, Trigger, FromStage, + /// ToStage, Taken, DeadEndReason, ReducerVersion, optional + /// nested ClassifierVerdict { ClassifierId, Level }). + /// + /// + public sealed class TelemetryTransitionEmitter + { + private static readonly JsonSerializerSettings WireFormatSettings = new JsonSerializerSettings + { + Converters = { new StringEnumConverter() }, + NullValueHandling = NullValueHandling.Ignore, + }; + + private readonly ITelemetryTransport _transport; + private readonly string _partitionKey; + + public TelemetryTransitionEmitter(ITelemetryTransport transport, string sessionId, string tenantId) + { + if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("SessionId is mandatory.", nameof(sessionId)); + if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId is mandatory.", nameof(tenantId)); + _transport = transport ?? throw new ArgumentNullException(nameof(transport)); + _partitionKey = $"{tenantId}_{sessionId}"; + } + + /// + /// Serialises the transition and enqueues it as a + /// draft. RowKey is the 10-digit + /// zero-padded — matches + /// TableDecisionTransitionRepository.BuildRowKey on the backend. + /// + public TelemetryItem Emit(DecisionTransition transition) + { + if (transition == null) throw new ArgumentNullException(nameof(transition)); + + var payloadJson = JsonConvert.SerializeObject(transition, Formatting.None, WireFormatSettings); + var rowKey = transition.StepIndex.ToString("D10"); + + var draft = new TelemetryItemDraft( + kind: TelemetryItemKind.DecisionTransition, + partitionKey: _partitionKey, + rowKey: rowKey, + payloadJson: payloadJson, + isSessionScoped: true, + requiresImmediateFlush: false); + + return _transport.Enqueue(draft); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryWireRoundTripTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryWireRoundTripTests.cs new file mode 100644 index 00000000..1f218f4c --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryWireRoundTripTests.cs @@ -0,0 +1,206 @@ +using System; +using System.Collections.Generic; +using AutopilotMonitor.DecisionCore.Classifiers; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Signals; +using AutopilotMonitor.DecisionCore.State; +using AutopilotMonitor.Functions.Functions.Ingest; +using AutopilotMonitor.Shared.Models; +using Newtonsoft.Json; +using Newtonsoft.Json.Converters; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Cross-layer round-trip (Codex Finding 1 follow-up gate): proves that a +/// / serialized on the V2 agent +/// deserializes correctly on the backend into a / +/// . Failure here = the agent's uploaded Signals or +/// Transitions never reach the Inspector / index pipeline intact. +/// +/// This test replicates the JsonSerializerSettings the agent-side emitters use +/// (Newtonsoft + + ). +/// The backend project can't reference the net48 agent assembly directly (one-way cross-TFM +/// constraint), so we exercise the wire contract via the shared DecisionCore types +/// plus the agreed serialization settings. Any drift between agent and backend would break +/// this assertion immediately. +/// +/// +public class TelemetryWireRoundTripTests +{ + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + + private static readonly DateTime At = new(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc); + + // Must mirror TelemetrySignalEmitter / TelemetryTransitionEmitter in V2.Core. + private static readonly JsonSerializerSettings AgentWireFormat = new() + { + Converters = { new StringEnumConverter() }, + NullValueHandling = NullValueHandling.Ignore, + }; + + private static TelemetryItemDto DraftToDto(string kind, string payloadJson, long sessionTraceOrdinal) => + new TelemetryItemDto + { + Kind = kind, + PartitionKey = $"{TenantId}_{SessionId}", + RowKey = "0", + TelemetryItemId = 1, + SessionTraceOrdinal = sessionTraceOrdinal, + PayloadJson = payloadJson, + EnqueuedAtUtc = At, + RetryCount = 0, + }; + + // ============================================================ Signal wire round-trip + + [Fact] + public void DecisionSignal_serialized_with_agent_settings_parses_into_SignalRecord() + { + var signal = new DecisionSignal( + sessionSignalOrdinal: 42, + sessionTraceOrdinal: 142, + kind: DecisionSignalKind.EspTerminalFailure, + kindSchemaVersion: 2, + occurredAtUtc: At, + sourceOrigin: "EspAndHelloTrackerAdapter", + evidence: new Evidence(EvidenceKind.Raw, "raw-42", "evidence-42")); + + var payloadJson = JsonConvert.SerializeObject(signal, AgentWireFormat); + var dto = DraftToDto("Signal", payloadJson, sessionTraceOrdinal: 142); + + var record = TelemetryPayloadParser.ParseSignal(dto, TenantId, SessionId); + + Assert.NotNull(record); + Assert.Equal(42L, record!.SessionSignalOrdinal); + Assert.Equal(142L, record.SessionTraceOrdinal); + Assert.Equal("EspTerminalFailure", record.Kind); + Assert.Equal(2, record.KindSchemaVersion); + Assert.Equal(At, record.OccurredAtUtc.ToUniversalTime()); + Assert.Equal("EspAndHelloTrackerAdapter", record.SourceOrigin); + Assert.Equal(payloadJson, record.PayloadJson); + } + + [Fact] + public void DecisionSignal_with_explicit_ordinal_zero_round_trips() + { + // Defense against the Finding 3 regression path: ordinal 0 is a legal value and must + // survive serialize → parse. (The parser's guard rejects *missing* ordinals, not + // legitimate 0s.) + var signal = new DecisionSignal( + sessionSignalOrdinal: 0, + sessionTraceOrdinal: 0, + kind: DecisionSignalKind.SessionStarted, + kindSchemaVersion: 1, + occurredAtUtc: At, + sourceOrigin: "SignalIngress", + evidence: new Evidence(EvidenceKind.Raw, "s", "start")); + + var payloadJson = JsonConvert.SerializeObject(signal, AgentWireFormat); + var record = TelemetryPayloadParser.ParseSignal( + DraftToDto("Signal", payloadJson, 0), TenantId, SessionId); + + Assert.NotNull(record); + Assert.Equal(0L, record!.SessionSignalOrdinal); + } + + // ============================================================ DecisionTransition wire round-trip + + [Fact] + public void DecisionTransition_Taken_terminal_with_classifier_round_trips_all_index_discriminators() + { + var verdict = new ClassifierVerdict( + classifierId: "whiteglove-sealing", + level: HypothesisLevel.Strong, + score: 80, + contributingFactors: new List { "PatternMatched" }, + reason: "matched sealing pattern", + inputHash: "abc123"); + + var transition = new DecisionTransition( + stepIndex: 7, + sessionTraceOrdinal: 207, + signalOrdinalRef: 42, + occurredAtUtc: At, + trigger: "EspExiting", + fromStage: SessionStage.EspDeviceSetup, + toStage: SessionStage.WhiteGloveSealed, + taken: true, + deadEndReason: null, + reducerVersion: "1.0.0", + classifierVerdict: verdict); + + var payloadJson = JsonConvert.SerializeObject(transition, AgentWireFormat); + var dto = DraftToDto("DecisionTransition", payloadJson, sessionTraceOrdinal: 207); + + var record = TelemetryPayloadParser.ParseTransition(dto, TenantId, SessionId); + + Assert.NotNull(record); + Assert.Equal(7, record!.StepIndex); + Assert.Equal(207L, record.SessionTraceOrdinal); + Assert.Equal(42L, record.SignalOrdinalRef); + Assert.Equal("EspExiting", record.Trigger); + Assert.Equal("EspDeviceSetup", record.FromStage); + Assert.Equal("WhiteGloveSealed", record.ToStage); + Assert.True(record.Taken); + Assert.True(record.IsTerminal); // Projected from ToStage + Assert.Equal("whiteglove-sealing", record.ClassifierVerdictId); + Assert.Equal("Strong", record.ClassifierHypothesisLevel); // StringEnumConverter → "Strong" + Assert.Equal("1.0.0", record.ReducerVersion); + Assert.Equal(payloadJson, record.PayloadJson); + } + + [Fact] + public void DecisionTransition_dead_end_preserves_DeadEndReason_and_drops_classifier_fields() + { + var blocked = new DecisionTransition( + stepIndex: 3, + sessionTraceOrdinal: 103, + signalOrdinalRef: 5, + occurredAtUtc: At, + trigger: "EspExiting", + fromStage: SessionStage.EspDeviceSetup, + toStage: SessionStage.EspDeviceSetup, + taken: false, + deadEndReason: "hybrid_reboot_gate_blocking", + reducerVersion: "1.0.0"); + + var payloadJson = JsonConvert.SerializeObject(blocked, AgentWireFormat); + var record = TelemetryPayloadParser.ParseTransition( + DraftToDto("DecisionTransition", payloadJson, 103), TenantId, SessionId); + + Assert.NotNull(record); + Assert.False(record!.Taken); + Assert.Equal("hybrid_reboot_gate_blocking", record.DeadEndReason); + Assert.False(record.IsTerminal); + Assert.Null(record.ClassifierVerdictId); + Assert.Null(record.ClassifierHypothesisLevel); + } + + [Fact] + public void DecisionTransition_with_step_zero_round_trips() + { + // StepIndex=0 is the first reducer step per session. Must survive serialize → parse + // (Finding 3 guard rejects missing StepIndex, not legitimate 0). + var first = new DecisionTransition( + stepIndex: 0, + sessionTraceOrdinal: 100, + signalOrdinalRef: 0, + occurredAtUtc: At, + trigger: "SessionStarted", + fromStage: SessionStage.Unknown, + toStage: SessionStage.AwaitingEspPhaseChange, + taken: true, + deadEndReason: null, + reducerVersion: "1.0.0"); + + var payloadJson = JsonConvert.SerializeObject(first, AgentWireFormat); + var record = TelemetryPayloadParser.ParseTransition( + DraftToDto("DecisionTransition", payloadJson, 100), TenantId, SessionId); + + Assert.NotNull(record); + Assert.Equal(0, record!.StepIndex); + Assert.Equal("AwaitingEspPhaseChange", record.ToStage); + } +}