feat(agent): signal-pipeline health block in agent_metrics_snapshot

The 2026-09-01 WG-seal soak on post-SignalIngress-fix builds (memory:
project_wg_seal_postfix_soak_residual) found sessions whose DecisionEngine
never sealed Part 1 while the agent kept running for hours - and telemetry
could not distinguish a frozen ingress worker from a deliberate Weak
sealing-classifier verdict, because the metrics snapshots carried no
pipeline counters at all.

Every agent_metrics_snapshot now carries the block. The fields are MONOTONIC
totals plus a monotonic queue peak - deliberately no momentary queue length:
a snapshot samples the integral, so any two snapshots prove whether the
worker ran in between (a frozen worker keeps signal_pending_count growing
while processed/applied stand still; a momentary gauge would alias between
ticks). Same field pattern the snapshot already uses for the spool
(spool_total_enqueued_count / spool_peak_pending_item_count).

- signal_last_ordinal / signal_processed_total (new ProcessedSignalCount) /
  signal_pending_count / signal_reentrant_total: SignalIngress worker
  progress; signal_queue_peak is a new lock-free Interlocked-max recorded in
  Post (fast + slow path; worker-local follow-up items deliberately excluded)
- decision_stage + decision_last_applied_ordinal: from the immutable
  DecisionState - answers the WG question directly ("minutes after the
  tracker's whiteglove_complete and still not WhiteGloveSealed")
- wg_sealing_level / wg_sealing_score: only after the sealing classifier
  produced at least one verdict (key ABSENCE, not Unknown/0 noise, for every
  non-WG session) - separates "classifier said Weak" from "never ran"

Wiring is a read-only probe chain following the imeTrackerHealthProbe
pattern: EnrollmentOrchestrator passes () => _processor?.CurrentState into
IComponentFactory.CreateCollectorHosts (new optional decisionStateProbe;
test fakes updated), PeriodicCollectorLifecycleHost combines it with its
existing concrete-SignalIngress reference in BuildSignalPipelineProbe, and
AgentSelfMetricsCollector writes the fields fail-soft (a throwing probe
degrades to a snapshot without the block - pinned by test). The snapshot
event travels the collector/event path, which is independent of the signal
pipeline it measures - field-proven by the freeze cases, whose perf/metrics
events kept flowing while the engine stood still.

No backend/web/MCP change needed: no new event type, DataJson is free-form.

Tests: 5 new collector facts (full block, absence semantics x2, fail-soft,
no-probe compatibility) + 2 new SignalIngress facts (processed total stable
across drain; queue peak survives the queue draining - the property a
momentary gauge would not have). Agent suite 2812 green, solution builds
with 0 warnings.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-09-01 18:26:22 +02:00
parent d763a885e2
commit 52ae6541a7
13 changed files with 416 additions and 12 deletions
+15
View File
@@ -251,6 +251,21 @@ Wired by `EnrollmentOrchestrator.Start` (`src/Agent/AutopilotMonitor.Agent.V2.Co
one-shot terminal-stage hook (raises the agent's `Terminated` event off-thread).
- The `DeadlineScheduler`'s `Fired` event loops back in as synthetic `DeadlineFired`
signals.
- **Pipeline health in `agent_metrics_snapshot`** (2026-09-01, motivated by the WG-seal
soak in which a frozen engine and a deliberate Weak sealing verdict were telemetrically
indistinguishable): every snapshot carries monotonic totals from `SignalIngress`
(`signal_last_ordinal`, `signal_processed_total`, `signal_pending_count`,
`signal_reentrant_total`, `signal_queue_peak` — a monotonic peak, deliberately no
momentary queue length, so snapshot sampling cannot alias it) plus `decision_stage`,
`decision_last_applied_ordinal` and — once the sealing classifier has produced a
verdict — `wg_sealing_level`/`wg_sealing_score` from the immutable `DecisionState`.
Two snapshots therefore prove whether the worker ran in between: a frozen worker keeps
`signal_pending_count` growing while the processed/applied counters stand still. The
probe chain is read-only (`EnrollmentOrchestrator` → `IComponentFactory.
CreateCollectorHosts(decisionStateProbe)` → `PeriodicCollectorLifecycleHost` →
`AgentSelfMetricsCollector`); the snapshot event itself travels the collector/event
path, which is independent of the signal pipeline it measures — field-proven by the
freeze cases, whose perf/metrics events kept flowing while the engine stood still.
- On restart, `RecoveryCoordinator` folds the persisted signal log through a transient
`new DecisionEngine()` before the live pipeline starts — see
[logs & persistence](logs-and-persistence.md).
+4
View File
@@ -1,5 +1,9 @@
# Log
## 2026-09-01 (3)
* **Update**: `agent/decision-engine.md` - agent_metrics_snapshot gains a signal-pipeline health block: monotonic SignalIngress totals (last ordinal, processed, pending, reentrant, queue peak) + decision_stage / decision_last_applied_ordinal / wg_sealing_level+score from the immutable DecisionState, wired via a read-only decisionStateProbe through IComponentFactory. Motivated by the WG-seal soak (project_wg_seal_postfix_soak_residual): a frozen engine and a deliberate Weak sealing verdict were previously indistinguishable in telemetry.
## 2026-09-01 (4)
* **Update**: `drift-guards.md` — MCP tool vocabularies are generated, not retyped. The codegen now emits `wire-vocabularies.generated.ts` (VALUES: session statuses, event severities, ops categories/severities/types, annotation lanes/verdicts) alongside the MCP's wire types, byte-pinned by the MCP freshness suite and the `shared-manifests-in-sync` CI job; the tool schemas derive their `z.enum()` from it. Found by the switch: `search_sessions` had offered 5 of 8 session statuses (no `AwaitingUser`/`Incomplete`) and both event readers 4 of 6 severities (no `Debug`/`Trace`). `vocabulary-drift.guard.test.ts` ratchets the remaining inline enums against a reasoned baseline. New `OpsEventTypes` (Shared) gives the 77 ops event types a canonical home — `OpsEventService` write sites use the constants (a raw literal now fails the dual-register test), and `get_resource(name="ops_event_types")` serves the vocabulary to the model.
@@ -0,0 +1,150 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using AutopilotMonitor.Agent.V2.Core.Logging;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Transport;
using AutopilotMonitor.Agent.V2.Core.Orchestration;
using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
using AutopilotMonitor.Agent.V2.Core.Tests.Orchestration;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.Shared;
using Xunit;
namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Telemetry
{
/// <summary>
/// Pins the signal-pipeline block of <c>agent_metrics_snapshot</c> (2026-09-01 WG-seal
/// soak: sessions whose DecisionEngine never sealed were telemetrically
/// indistinguishable from a deliberate Weak classifier verdict, because the snapshots
/// carried no pipeline counters). The fields are monotonic totals plus a monotonic
/// queue peak — deliberately no momentary queue length — so two snapshots prove
/// whether the ingress worker ran in between.
/// </summary>
public sealed class AgentSelfMetricsCollectorSignalPipelineTests
{
private static AgentLogger NewLogger(string path)
=> new AgentLogger(Path.Combine(path, "logs"), AgentLogLevel.Info);
private static (AgentSelfMetricsCollector Collector, FakeSignalIngressSink Sink) Build(
string tmpPath, Func<SignalPipelineHealth?>? probe)
{
var sink = new FakeSignalIngressSink();
var post = new InformationalEventPost(sink, SystemClock.Instance);
var collector = new AgentSelfMetricsCollector(
sessionId: "S1",
tenantId: "T1",
post: post,
networkMetrics: new NetworkMetrics(),
logger: NewLogger(tmpPath),
agentVersion: "2.0.test",
signalPipelineProbe: probe);
return (collector, sink);
}
private static IDictionary<string, object> CollectSnapshotData(
AgentSelfMetricsCollector collector, FakeSignalIngressSink sink)
{
collector.CollectSafe();
var snapshot = sink.Posted.Single(p =>
p.TypedPayload is IDictionary<string, object> d
&& d.ContainsKey("agent_version"));
return (IDictionary<string, object>)snapshot.TypedPayload!;
}
[Fact]
public void Snapshot_carries_pipeline_counters_stage_and_wg_verdict()
{
using var tmp = new TempDirectory();
var (collector, sink) = Build(tmp.Path, () => new SignalPipelineHealth
{
LastAssignedSignalOrdinal = 41,
ProcessedSignalCount = 42,
PendingSignalCount = 3,
ReentrantPostCount = 7,
QueueLengthPeak = 12,
DecisionStage = "EspDeviceSetup",
LastAppliedSignalOrdinal = 40,
WhiteGloveSealingLevel = "Weak",
WhiteGloveSealingScore = 40,
});
var data = CollectSnapshotData(collector, sink);
Assert.Equal(41L, data["signal_last_ordinal"]);
Assert.Equal(42L, data["signal_processed_total"]);
Assert.Equal(3L, data["signal_pending_count"]);
Assert.Equal(7L, data["signal_reentrant_total"]);
Assert.Equal(12L, data["signal_queue_peak"]);
Assert.Equal("EspDeviceSetup", data["decision_stage"]);
Assert.Equal(40L, data["decision_last_applied_ordinal"]);
Assert.Equal("Weak", data["wg_sealing_level"]);
Assert.Equal(40, data["wg_sealing_score"]);
}
[Fact]
public void State_derived_fields_are_omitted_when_the_probe_has_no_state()
{
// Counters-only health (test fakes without a decision-state probe): the numeric
// block still lands, the state-derived keys must be ABSENT — not null-valued.
using var tmp = new TempDirectory();
var (collector, sink) = Build(tmp.Path, () => new SignalPipelineHealth
{
ProcessedSignalCount = 5,
});
var data = CollectSnapshotData(collector, sink);
Assert.Equal(5L, data["signal_processed_total"]);
Assert.False(data.ContainsKey("decision_stage"));
Assert.False(data.ContainsKey("decision_last_applied_ordinal"));
Assert.False(data.ContainsKey("wg_sealing_level"));
Assert.False(data.ContainsKey("wg_sealing_score"));
}
[Fact]
public void Wg_fields_are_omitted_before_the_classifier_produced_a_verdict()
{
// Level stays null until the classifier ran — every non-WG session would
// otherwise carry a meaningless Unknown/0 pair on each snapshot.
using var tmp = new TempDirectory();
var (collector, sink) = Build(tmp.Path, () => new SignalPipelineHealth
{
DecisionStage = "Monitoring",
LastAppliedSignalOrdinal = 9,
});
var data = CollectSnapshotData(collector, sink);
Assert.Equal("Monitoring", data["decision_stage"]);
Assert.False(data.ContainsKey("wg_sealing_level"));
Assert.False(data.ContainsKey("wg_sealing_score"));
}
[Fact]
public void Probe_failure_degrades_to_a_snapshot_without_the_block()
{
using var tmp = new TempDirectory();
var (collector, sink) = Build(tmp.Path,
() => throw new InvalidOperationException("probe exploded"));
var data = CollectSnapshotData(collector, sink);
Assert.True(data.ContainsKey("agent_version")); // snapshot itself survives
Assert.DoesNotContain(data.Keys, k => k.StartsWith("signal_", StringComparison.Ordinal));
}
[Fact]
public void No_probe_means_no_pipeline_fields()
{
using var tmp = new TempDirectory();
var (collector, sink) = Build(tmp.Path, probe: null);
var data = CollectSnapshotData(collector, sink);
Assert.DoesNotContain(data.Keys, k => k.StartsWith("signal_", StringComparison.Ordinal));
Assert.False(data.ContainsKey("decision_stage"));
}
}
}
@@ -64,7 +64,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
ISignalIngressSink ingress,
IClock clock,
AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.ITelemetrySpool? telemetrySpool,
TimelineEventStream? timelineEvents = null)
TimelineEventStream? timelineEvents = null,
Func<AutopilotMonitor.DecisionCore.State.DecisionState?>? decisionStateProbe = null)
{
CapturedWhiteGloveSealingPatternIds = whiteGloveSealingPatternIds;
CapturedIngress = ingress;
@@ -28,7 +28,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
ISignalIngressSink ingress,
IClock clock,
AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.ITelemetrySpool? telemetrySpool,
TimelineEventStream? timelineEvents = null)
TimelineEventStream? timelineEvents = null,
Func<AutopilotMonitor.DecisionCore.State.DecisionState?>? decisionStateProbe = null)
{
CapturedPatternIds = whiteGloveSealingPatternIds;
return new CollectorSurfaces(Array.Empty<ICollectorHost>());
@@ -65,6 +65,50 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
return SpinWait.SpinUntil(condition, timeoutMs);
}
// ==================================== Pipeline-health counters (metrics snapshot)
// The agent_metrics_snapshot exports these as MONOTONIC totals so two snapshots can
// prove whether the worker ran in between (2026-09-01 WG-seal soak: a frozen engine
// and a deliberate Weak sealing verdict were otherwise indistinguishable).
[Fact]
public void ProcessedSignalCount_counts_fully_processed_signals()
{
using var rig = new Rig();
var ing = rig.Build();
Assert.Equal(0, ing.ProcessedSignalCount);
ing.Start();
for (int i = 0; i < 5; i++)
ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence($"p-{i}"));
Assert.True(WaitFor(() => ing.ProcessedSignalCount == 5),
$"expected 5 processed, saw {ing.ProcessedSignalCount}");
Assert.Equal(0, ing.PendingSignalCount);
ing.Stop();
Assert.Equal(5, ing.ProcessedSignalCount); // stable after drain
}
[Fact]
public void QueueLengthPeak_is_monotonic_and_survives_drain()
{
using var rig = new Rig();
var ing = rig.Build();
// Post BEFORE Start: the worker is not consuming yet, so the channel depth
// deterministically reaches the number of queued items.
for (int i = 0; i < 4; i++)
ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence($"q-{i}"));
Assert.True(ing.QueueLengthPeak >= 4, $"peak {ing.QueueLengthPeak} < 4");
ing.Start();
Assert.True(WaitFor(() => ing.ProcessedSignalCount == 4));
// Drained queue must NOT reset the peak — that is the whole point of exporting
// a peak instead of a momentary length (snapshot sampling cannot alias it away).
Assert.True(ing.QueueLengthPeak >= 4);
ing.Stop();
}
// ============================================================ Lifecycle
[Fact]
@@ -34,6 +34,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
private readonly NetworkMetrics _networkMetrics;
private readonly ITelemetrySpool? _telemetrySpool;
private readonly Func<Monitoring.Enrollment.Ime.ImeTrackerHealth?>? _imeTrackerHealthProbe;
private readonly Func<Orchestration.SignalPipelineHealth?>? _signalPipelineProbe;
// Previous sample for delta calculations
private TimeSpan _prevCpuTime;
@@ -53,13 +54,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
string agentVersion = "unknown",
int intervalSeconds = 60,
ITelemetrySpool? telemetrySpool = null,
Func<Monitoring.Enrollment.Ime.ImeTrackerHealth?>? imeTrackerHealthProbe = null)
Func<Monitoring.Enrollment.Ime.ImeTrackerHealth?>? imeTrackerHealthProbe = null,
Func<Orchestration.SignalPipelineHealth?>? signalPipelineProbe = null)
: base(sessionId, tenantId, post, logger, intervalSeconds)
{
_networkMetrics = networkMetrics ?? throw new ArgumentNullException(nameof(networkMetrics));
_agentVersion = string.IsNullOrWhiteSpace(agentVersion) ? "unknown" : agentVersion;
_telemetrySpool = telemetrySpool;
_imeTrackerHealthProbe = imeTrackerHealthProbe;
_signalPipelineProbe = signalPipelineProbe;
}
protected override void OnBeforeStart()
@@ -82,11 +85,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
protected override void Collect()
{
// Full path writes 29 keys: agent_version + process metrics (5: cpu, ws, private,
// Full path writes 38 keys: agent_version + process metrics (5: cpu, ws, private,
// threads, handles) + spool stats (5) + network delta (9: requests, failures,
// bytes_up/down, avg_latency, total_up/down/requests/latency) + IME tracker health
// (9). cap=29 → HashHelpers.GetPrime(29)=29 buckets → no resize on the last key.
var data = new Dictionary<string, object>(capacity: 29, StringComparer.Ordinal)
// (9) + signal pipeline (9: 5 counters, stage, applied ordinal, wg level+score).
// cap=38 → HashHelpers.GetPrime(38)=41 buckets → no resize on the last key.
var data = new Dictionary<string, object>(capacity: 38, StringComparer.Ordinal)
{
{ "agent_version", _agentVersion }
};
@@ -213,6 +217,44 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
}
}
// --- Signal/decision pipeline health ---
// Monotonic totals + a monotonic queue peak, deliberately NO momentary queue
// length: a snapshot samples the integral, so any two snapshots prove whether
// the ingress worker ran in between. Field motivation (2026-09-01 WG-seal soak):
// sessions whose engine never sealed were indistinguishable from a deliberate
// Weak classifier verdict without these fields. decision_stage answers the WG
// question directly; the wg_sealing_* pair separates "classifier said Weak"
// from "classifier never ran".
if (_signalPipelineProbe != null)
{
try
{
var pipeline = _signalPipelineProbe();
if (pipeline != null)
{
data["signal_last_ordinal"] = pipeline.LastAssignedSignalOrdinal;
data["signal_processed_total"] = pipeline.ProcessedSignalCount;
data["signal_pending_count"] = pipeline.PendingSignalCount;
data["signal_reentrant_total"] = pipeline.ReentrantPostCount;
data["signal_queue_peak"] = pipeline.QueueLengthPeak;
if (pipeline.DecisionStage != null)
data["decision_stage"] = pipeline.DecisionStage;
if (pipeline.LastAppliedSignalOrdinal.HasValue)
data["decision_last_applied_ordinal"] = pipeline.LastAppliedSignalOrdinal.Value;
if (pipeline.WhiteGloveSealingLevel != null)
{
data["wg_sealing_level"] = pipeline.WhiteGloveSealingLevel;
if (pipeline.WhiteGloveSealingScore.HasValue)
data["wg_sealing_score"] = pipeline.WhiteGloveSealingScore.Value;
}
}
}
catch (Exception ex)
{
Logger.Debug($"Signal pipeline health read failed: {ex.Message}");
}
}
if (data.Count > 0)
{
Post.Emit(new EnrollmentEvent
@@ -204,7 +204,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
ISignalIngressSink ingress,
IClock clock,
Transport.Telemetry.ITelemetrySpool? telemetrySpool,
TimelineEventStream? timelineEvents = null)
TimelineEventStream? timelineEvents = null,
Func<AutopilotMonitor.DecisionCore.State.DecisionState?>? decisionStateProbe = null)
{
if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("SessionId required.", nameof(sessionId));
if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId required.", nameof(tenantId));
@@ -599,7 +600,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
startupGate: _startupEventGate, // M3 — disk_space_low latch survives restarts
// IME tracker health rides the metrics snapshot (queue depth + skip counters);
// read-only probe, same shape as the registry observer's trackerStateProbe.
imeTrackerHealthProbe: () => imeLogHostRef?.GetTrackerHealth()));
imeTrackerHealthProbe: () => imeLogHostRef?.GetTrackerHealth(),
// Stage + WG classifier verdict for the snapshot's signal-pipeline block
// (2026-09-01 WG-seal soak: without them a frozen engine and a deliberate
// Weak sealing verdict were telemetrically indistinguishable).
decisionStateProbe: decisionStateProbe));
}
// V1 parity (CollectorCoordinator.StartOptionalCollectors:375-382) — wire the
@@ -667,7 +667,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
var surfaces = _componentFactory.CreateCollectorHosts(
_sessionId, _tenantId, _logger, _whiteGloveSealingPatternIds,
ingress: _ingress, clock: _clock, telemetrySpool: _spool,
timelineEvents: _timelineEvents);
timelineEvents: _timelineEvents,
// Immutable-state reference read; null-safe against a processor that is
// torn down while a late metrics tick still fires.
decisionStateProbe: () => _processor?.CurrentState);
CollectorSurfaces = surfaces;
_collectorHosts = surfaces.Hosts;
foreach (var host in _collectorHosts)
@@ -8,6 +8,7 @@ using AutopilotMonitor.Agent.V2.Core.Monitoring.Transport;
using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using AutopilotMonitor.DecisionCore.State;
using AutopilotMonitor.Shared.Models;
using SharedConstants = AutopilotMonitor.Shared.Constants;
@@ -56,6 +57,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private readonly ITelemetrySpool? _telemetrySpool;
private readonly Persistence.StartupEventGate? _startupGate;
private readonly Func<Monitoring.Enrollment.Ime.ImeTrackerHealth?>? _imeTrackerHealthProbe;
// Read-only probe into the reducer's current DecisionState (stage + WG classifier
// verdict) for the signal-pipeline block of agent_metrics_snapshot. DecisionState is
// immutable, so a cross-thread reference read yields a consistent (possibly slightly
// stale) snapshot — same access pattern the orchestrator's own Stop path uses.
private readonly Func<DecisionState?>? _decisionStateProbe;
// Codex Finding 4 — reference to the concrete SignalIngress (when available) so we
// can subscribe to its SignalPosted event. This lets us observe activity from
@@ -90,9 +96,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
string agentVersion,
ITelemetrySpool? telemetrySpool = null,
Persistence.StartupEventGate? startupGate = null,
Func<Monitoring.Enrollment.Ime.ImeTrackerHealth?>? imeTrackerHealthProbe = null)
Func<Monitoring.Enrollment.Ime.ImeTrackerHealth?>? imeTrackerHealthProbe = null,
Func<DecisionState?>? decisionStateProbe = null)
{
_imeTrackerHealthProbe = imeTrackerHealthProbe;
_decisionStateProbe = decisionStateProbe;
if (ingress == null) throw new ArgumentNullException(nameof(ingress));
if (clock == null) throw new ArgumentNullException(nameof(clock));
_sessionId = sessionId;
@@ -198,7 +206,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
{
_selfMetricsCollector = new AgentSelfMetricsCollector(
_sessionId, _tenantId, _post, _networkMetrics, _logger, _agentVersion, _selfMetricsIntervalSeconds, _telemetrySpool,
imeTrackerHealthProbe: _imeTrackerHealthProbe);
imeTrackerHealthProbe: _imeTrackerHealthProbe,
signalPipelineProbe: BuildSignalPipelineProbe());
_selfMetricsCollector.Start();
}
_idleStopped = false;
@@ -215,6 +224,48 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
_selfMetricsCollector = null;
}
/// <summary>
/// Builds the signal-pipeline probe for the metrics snapshot from whatever is
/// available: the concrete <see cref="SignalIngress"/> supplies the monotonic
/// worker counters, the decision-state probe supplies stage + WG classifier
/// verdict. Null only when NEITHER source exists (pure test fakes) — then the
/// snapshot simply omits the block, same degrade pattern as the IME health probe.
/// </summary>
private Func<SignalPipelineHealth?>? BuildSignalPipelineProbe()
{
var ingress = _observableIngress;
var stateProbe = _decisionStateProbe;
if (ingress == null && stateProbe == null) return null;
return () =>
{
var health = new SignalPipelineHealth();
if (ingress != null)
{
health.LastAssignedSignalOrdinal = ingress.LastAssignedSignalOrdinal;
health.ProcessedSignalCount = ingress.ProcessedSignalCount;
health.PendingSignalCount = ingress.PendingSignalCount;
health.ReentrantPostCount = ingress.ReentrantPostCount;
health.QueueLengthPeak = ingress.QueueLengthPeak;
}
var state = stateProbe?.Invoke();
if (state != null)
{
health.DecisionStage = state.Stage.ToString();
health.LastAppliedSignalOrdinal = state.LastAppliedSignalOrdinal;
// Only after the classifier produced at least one verdict — otherwise
// every non-WG session would carry a meaningless Unknown/0 pair.
var sealing = state.ClassifierOutcomes?.WhiteGloveSealing;
if (sealing?.LastClassifierVerdictId != null)
{
health.WhiteGloveSealingLevel = sealing.Level.ToString();
health.WhiteGloveSealingScore = sealing.Score;
}
}
return health;
};
}
private void OnRealEvent()
{
lock (_sync)
@@ -1,4 +1,5 @@
#nullable enable
using System;
using System.Collections.Generic;
using AutopilotMonitor.Agent.V2.Core.Logging;
using AutopilotMonitor.DecisionCore.Engine;
@@ -47,6 +48,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// and the host's phase/event triggers degrade off (startup + interval rules still run).
/// </para>
/// </summary>
/// <remarks>
/// <paramref name="decisionStateProbe"/> is a read-only probe into the reducer's current
/// immutable <see cref="AutopilotMonitor.DecisionCore.State.DecisionState"/> — it feeds the
/// signal-pipeline block of <c>agent_metrics_snapshot</c> (stage + WG classifier verdict).
/// Nullable; fakes pass null and the snapshot omits the state-derived fields.
/// </remarks>
CollectorSurfaces CreateCollectorHosts(
string sessionId,
string tenantId,
@@ -55,6 +62,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
ISignalIngressSink ingress,
IClock clock,
Transport.Telemetry.ITelemetrySpool? telemetrySpool,
TimelineEventStream? timelineEvents = null);
TimelineEventStream? timelineEvents = null,
Func<AutopilotMonitor.DecisionCore.State.DecisionState?>? decisionStateProbe = null);
}
}
@@ -94,6 +94,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
// next channel item is taken. Only the worker enqueues and dequeues — no lock.
private readonly Queue<IngressItem> _workerLocalQueue = new Queue<IngressItem>();
private long _reentrantPostCount;
private long _queueLengthPeak;
private long _nextSignalOrdinal; // worker-thread-only after Start; seeded from SignalLog
private int _started; // 0 = not started, 1 = running
private int _stopRequested; // 0 = live, 1 = Stop called
@@ -156,6 +157,21 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// </summary>
public long ReentrantPostCount => Interlocked.Read(ref _reentrantPostCount);
/// <summary>
/// Total signals the worker has fully processed (reduce + effects). Monotonic — the
/// agent_metrics_snapshot exports this so two snapshots prove whether the pipeline
/// worked in between (a frozen worker keeps posting-side counters growing while this
/// one stands still; a momentary queue-length sample could never show that).
/// </summary>
public long ProcessedSignalCount => Interlocked.Read(ref _processedCount);
/// <summary>
/// Highest channel queue length observed at any Post since Start. Monotonic
/// non-decreasing, so snapshot sampling cannot alias it away (same rationale as the
/// spool's PeakPendingItemCount). Worker-local follow-up items never count here.
/// </summary>
public long QueueLengthPeak => Interlocked.Read(ref _queueLengthPeak);
/// <summary>
/// Fires synchronously from <see cref="Post"/> after a signal has been accepted into
/// the channel (for both the fast-path <see cref="BlockingCollection{T}.TryAdd"/> and
@@ -317,6 +333,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
if (added)
{
UpdateQueueLengthPeak(_channel.Count);
RaiseSignalPosted(kind, payload);
return;
}
@@ -324,6 +341,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
// Slow path: bounded channel full → block, measure duration, notify observer throttled.
var blockStartedUtc = _clock.UtcNow;
int queueLenAtBlock = _channel.Count;
UpdateQueueLengthPeak(queueLenAtBlock);
try
{
_channel.Add(item);
@@ -345,6 +363,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
RaiseSignalPosted(kind, payload);
}
/// <summary>Lock-free monotonic max (net48 has no Interlocked max primitive).</summary>
private void UpdateQueueLengthPeak(int observedLength)
{
long observed = observedLength;
long current = Interlocked.Read(ref _queueLengthPeak);
while (observed > current)
{
var prior = Interlocked.CompareExchange(ref _queueLengthPeak, observed, current);
if (prior == current) return;
current = prior;
}
}
private void RaiseSignalPosted(DecisionSignalKind kind, IReadOnlyDictionary<string, string>? payload)
{
// Snapshot the delegate once — standard pattern to tolerate concurrent subscribe /
@@ -0,0 +1,49 @@
#nullable enable annotations
using System;
namespace AutopilotMonitor.Agent.V2.Core.Orchestration
{
/// <summary>
/// Point-in-time read model of the signal/decision pipeline for the
/// <c>agent_metrics_snapshot</c> event. Motivated by the 2026-09-01 WG-seal soak
/// (memory: project_wg_seal_postfix_soak_residual): sessions whose DecisionEngine never
/// sealed were telemetrically indistinguishable from a deliberate Weak classifier verdict,
/// because the snapshots carried no signal-pipeline counters at all.
/// <para>
/// The counters are MONOTONIC totals, not momentary gauges — a snapshot samples the
/// integral, so any two snapshots prove whether the worker ran in between (a frozen
/// worker keeps <see cref="PendingSignalCount"/> growing while
/// <see cref="ProcessedSignalCount"/> / <see cref="LastAppliedSignalOrdinal"/> stand
/// still). Momentary queue length is deliberately exported only as a peak.
/// </para>
/// </summary>
public sealed class SignalPipelineHealth
{
/// <summary>Last SessionSignalOrdinal assigned by the ingress worker; −1 before the first signal.</summary>
public long LastAssignedSignalOrdinal { get; set; }
/// <summary>Total signals fully processed (reduce + effects). Monotonic.</summary>
public long ProcessedSignalCount { get; set; }
/// <summary>Signals accepted by Post but not yet fully processed (backlog gauge; pairs with the totals).</summary>
public long PendingSignalCount { get; set; }
/// <summary>Worker-thread re-entrant posts routed via the worker-local queue. Monotonic.</summary>
public long ReentrantPostCount { get; set; }
/// <summary>Highest channel queue length observed at any Post since Start. Monotonic non-decreasing.</summary>
public long QueueLengthPeak { get; set; }
/// <summary>Current DecisionState stage name (e.g. "EspDeviceSetup", "WhiteGloveSealed"). Null when no state is available.</summary>
public string? DecisionStage { get; set; }
/// <summary>The DecisionState's LastAppliedSignalOrdinal — the reducer-side twin of the ingress counters. Null when no state is available.</summary>
public long? LastAppliedSignalOrdinal { get; set; }
/// <summary>Last WhiteGlove-sealing classifier level ("Weak", "Confirmed", …). Null until the classifier produced a verdict.</summary>
public string? WhiteGloveSealingLevel { get; set; }
/// <summary>Score of the last WhiteGlove-sealing verdict. Null until the classifier produced a verdict.</summary>
public int? WhiteGloveSealingScore { get; set; }
}
}