mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 12:42:38 +00:00
fix(agent): upload V2 Signals + DecisionTransitions to backend (Codex Finding 1)
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) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
+147
@@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// Codex Finding 1 regression gate for <see cref="DecisionStepProcessor"/>: verifies that
|
||||
/// every successfully-journaled <see cref="DecisionTransition"/> is also projected onto
|
||||
/// the telemetry transport for backend upload, and that a throwing emitter does NOT
|
||||
/// abort the step.
|
||||
/// </summary>
|
||||
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<InvalidOperationException>(() => 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<DrainResult> DrainAllAsync(CancellationToken cancellationToken = default)
|
||||
=> Task.FromResult(DrainResult.Empty());
|
||||
|
||||
public void Dispose() { }
|
||||
}
|
||||
}
|
||||
}
|
||||
+157
@@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// Codex Finding 1 regression gate: without the emitter hook in
|
||||
/// <see cref="SignalIngress"/>, no signal ever reached the telemetry transport and the
|
||||
/// backend <c>/api/sessions/{id}/signals</c> 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.
|
||||
/// </summary>
|
||||
[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<bool> 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<DrainResult> DrainAllAsync(CancellationToken cancellationToken = default)
|
||||
=> System.Threading.Tasks.Task.FromResult(DrainResult.Empty());
|
||||
|
||||
public void Dispose() { }
|
||||
}
|
||||
}
|
||||
}
|
||||
+113
@@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// Pure emitter tests for <see cref="TelemetrySignalEmitter"/> (Codex Finding 1 fix).
|
||||
/// Verifies the agent-side wire format and transport projection; the cross-layer
|
||||
/// round-trip with the backend parser lives in <c>TelemetrySignalRoundTripTests</c>.
|
||||
/// </summary>
|
||||
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<ArgumentNullException>(() => emitter.Emit(null!));
|
||||
Assert.Equal(0, transport.EnqueueCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Ctor_rejects_null_or_empty_identifiers()
|
||||
{
|
||||
var transport = new FakeTelemetryTransport();
|
||||
Assert.Throws<ArgumentException>(() => new TelemetrySignalEmitter(transport, "", TenantId));
|
||||
Assert.Throws<ArgumentException>(() => new TelemetrySignalEmitter(transport, SessionId, ""));
|
||||
Assert.Throws<ArgumentNullException>(() => new TelemetrySignalEmitter(null!, SessionId, TenantId));
|
||||
}
|
||||
}
|
||||
}
|
||||
+168
@@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// Pure emitter tests for <see cref="TelemetryTransitionEmitter"/> (Codex Finding 1 fix).
|
||||
/// Cross-layer round-trip with the backend parser lives in
|
||||
/// <c>TelemetryTransitionRoundTripTests</c>.
|
||||
/// </summary>
|
||||
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<string> { "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<ArgumentNullException>(() => emitter.Emit(null!));
|
||||
Assert.Equal(0, transport.EnqueueCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Ctor_rejects_null_or_empty_identifiers()
|
||||
{
|
||||
var transport = new FakeTelemetryTransport();
|
||||
Assert.Throws<ArgumentException>(() => new TelemetryTransitionEmitter(transport, "", TenantId));
|
||||
Assert.Throws<ArgumentException>(() => new TelemetryTransitionEmitter(transport, SessionId, ""));
|
||||
Assert.Throws<ArgumentNullException>(() => new TelemetryTransitionEmitter(null!, SessionId, TenantId));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<DecisionState>? _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<DecisionState>? onTerminalStageReached = null)
|
||||
Action<DecisionState>? 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).
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// Emitter that projects <see cref="DecisionSignal"/>s onto the generic
|
||||
/// <see cref="ITelemetryTransport"/> for upload to the backend <c>/api/agent/telemetry</c>
|
||||
/// endpoint (Plan §2.7a, M5.a <c>Signals</c> table). Sibling of
|
||||
/// <see cref="Events.TelemetryEventEmitter"/>.
|
||||
/// <para>
|
||||
/// Called by <see cref="Orchestration.SignalIngress"/> after <c>SignalLog.Append</c> 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Wire format:</b> Newtonsoft PascalCase with <see cref="StringEnumConverter"/> for
|
||||
/// <see cref="DecisionSignalKind"/>. Matches the backend-side shape
|
||||
/// <c>TelemetryPayloadParser.ParseSignal</c> expects (<c>SessionSignalOrdinal</c>,
|
||||
/// <c>Kind</c>, <c>KindSchemaVersion</c>, <c>OccurredAtUtc</c>, <c>SourceOrigin</c>,
|
||||
/// <c>SessionTraceOrdinal</c>).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
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}";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Serialises the signal and enqueues it as a <see cref="TelemetryItemKind.Signal"/>
|
||||
/// draft. RowKey is the 19-digit zero-padded <see cref="DecisionSignal.SessionSignalOrdinal"/>
|
||||
/// — matches <c>TableSignalRepository.BuildRowKey</c> on the backend so Upsert collapses
|
||||
/// duplicates.
|
||||
/// </summary>
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
+72
@@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// Emitter that projects <see cref="DecisionTransition"/>s onto the generic
|
||||
/// <see cref="ITelemetryTransport"/> for upload to the backend <c>/api/agent/telemetry</c>
|
||||
/// endpoint (Plan §2.7a, M5.a <c>DecisionTransitions</c> table). Sibling of
|
||||
/// <see cref="Signals.TelemetrySignalEmitter"/> / <see cref="Events.TelemetryEventEmitter"/>.
|
||||
/// <para>
|
||||
/// Called by <see cref="Orchestration.DecisionStepProcessor"/> after
|
||||
/// <c>Journal.Append(transition)</c> has committed locally — Journal is the authoritative
|
||||
/// store (§2.7c / L.1), upload is a best-effort projection.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Wire format:</b> Newtonsoft PascalCase with <see cref="StringEnumConverter"/> for
|
||||
/// <c>SessionStage</c>, <c>HypothesisLevel</c>, etc. Matches the backend shape
|
||||
/// <c>TelemetryPayloadParser.ParseTransition</c> expects (<c>StepIndex</c>,
|
||||
/// <c>SessionTraceOrdinal</c>, <c>SignalOrdinalRef</c>, <c>Trigger</c>, <c>FromStage</c>,
|
||||
/// <c>ToStage</c>, <c>Taken</c>, <c>DeadEndReason</c>, <c>ReducerVersion</c>, optional
|
||||
/// nested <c>ClassifierVerdict { ClassifierId, Level }</c>).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
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}";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Serialises the transition and enqueues it as a
|
||||
/// <see cref="TelemetryItemKind.DecisionTransition"/> draft. RowKey is the 10-digit
|
||||
/// zero-padded <see cref="DecisionTransition.StepIndex"/> — matches
|
||||
/// <c>TableDecisionTransitionRepository.BuildRowKey</c> on the backend.
|
||||
/// </summary>
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// <b>Cross-layer round-trip</b> (Codex Finding 1 follow-up gate): proves that a
|
||||
/// <see cref="DecisionSignal"/> / <see cref="DecisionTransition"/> serialized on the V2 agent
|
||||
/// deserializes correctly on the backend into a <see cref="SignalRecord"/> /
|
||||
/// <see cref="DecisionTransitionRecord"/>. Failure here = the agent's uploaded Signals or
|
||||
/// Transitions never reach the Inspector / index pipeline intact.
|
||||
/// <para>
|
||||
/// This test replicates the <c>JsonSerializerSettings</c> the agent-side emitters use
|
||||
/// (Newtonsoft + <see cref="StringEnumConverter"/> + <see cref="NullValueHandling.Ignore"/>).
|
||||
/// 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 <c>DecisionCore</c> types
|
||||
/// plus the agreed serialization settings. Any drift between agent and backend would break
|
||||
/// this assertion immediately.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
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<string> { "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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user