diff --git a/docs/agent/decision-engine.md b/docs/agent/decision-engine.md index c3bf2bf1..d9aa99a8 100644 --- a/docs/agent/decision-engine.md +++ b/docs/agent/decision-engine.md @@ -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). diff --git a/docs/log.md b/docs/log.md index 67dc600d..e5177fed 100644 --- a/docs/log.md +++ b/docs/log.md @@ -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. diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Telemetry/AgentSelfMetricsCollectorSignalPipelineTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Telemetry/AgentSelfMetricsCollectorSignalPipelineTests.cs new file mode 100644 index 00000000..afd5196e --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Telemetry/AgentSelfMetricsCollectorSignalPipelineTests.cs @@ -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 +{ + /// + /// Pins the signal-pipeline block of agent_metrics_snapshot (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. + /// + 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? 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 CollectSnapshotData( + AgentSelfMetricsCollector collector, FakeSignalIngressSink sink) + { + collector.CollectSafe(); + var snapshot = sink.Posted.Single(p => + p.TypedPayload is IDictionary d + && d.ContainsKey("agent_version")); + return (IDictionary)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")); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs index 25f7e8df..ceb97d3f 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs @@ -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? decisionStateProbe = null) { CapturedWhiteGloveSealingPatternIds = whiteGloveSealingPatternIds; CapturedIngress = ingress; diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorWhiteGlovePatternTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorWhiteGlovePatternTests.cs index 030d0039..1f3738b7 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorWhiteGlovePatternTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorWhiteGlovePatternTests.cs @@ -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? decisionStateProbe = null) { CapturedPatternIds = whiteGloveSealingPatternIds; return new CollectorSurfaces(Array.Empty()); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs index 9352c09e..7256985a 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs @@ -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] diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Periodic/AgentSelfMetricsCollector.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Periodic/AgentSelfMetricsCollector.cs index 7ce5baaa..428bdc8a 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Periodic/AgentSelfMetricsCollector.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Periodic/AgentSelfMetricsCollector.cs @@ -34,6 +34,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic private readonly NetworkMetrics _networkMetrics; private readonly ITelemetrySpool? _telemetrySpool; private readonly Func? _imeTrackerHealthProbe; + private readonly Func? _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? imeTrackerHealthProbe = null) + Func? imeTrackerHealthProbe = null, + Func? 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(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(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 diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs index 7a06e069..65e13672 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs @@ -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? 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 diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs index 04a1c520..b113884d 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs @@ -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) diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/PeriodicCollectorLifecycleHost.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/PeriodicCollectorLifecycleHost.cs index b43d7a5e..971e264c 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/PeriodicCollectorLifecycleHost.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/PeriodicCollectorLifecycleHost.cs @@ -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? _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? _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? imeTrackerHealthProbe = null) + Func? imeTrackerHealthProbe = null, + Func? 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; } + /// + /// Builds the signal-pipeline probe for the metrics snapshot from whatever is + /// available: the concrete 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. + /// + private Func? 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) diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs index 19d3aece..48ea6f1a 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs @@ -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). /// /// + /// + /// is a read-only probe into the reducer's current + /// immutable — it feeds the + /// signal-pipeline block of agent_metrics_snapshot (stage + WG classifier verdict). + /// Nullable; fakes pass null and the snapshot omits the state-derived fields. + /// 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? decisionStateProbe = null); } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs index d9df960a..bba48e06 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs @@ -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 _workerLocalQueue = new Queue(); 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 /// public long ReentrantPostCount => Interlocked.Read(ref _reentrantPostCount); + /// + /// 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). + /// + public long ProcessedSignalCount => Interlocked.Read(ref _processedCount); + + /// + /// 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. + /// + public long QueueLengthPeak => Interlocked.Read(ref _queueLengthPeak); + /// /// Fires synchronously from after a signal has been accepted into /// the channel (for both the fast-path 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); } + /// Lock-free monotonic max (net48 has no Interlocked max primitive). + 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? payload) { // Snapshot the delegate once — standard pattern to tolerate concurrent subscribe / diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalPipelineHealth.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalPipelineHealth.cs new file mode 100644 index 00000000..61fc14a6 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalPipelineHealth.cs @@ -0,0 +1,49 @@ +#nullable enable annotations +using System; + +namespace AutopilotMonitor.Agent.V2.Core.Orchestration +{ + /// + /// Point-in-time read model of the signal/decision pipeline for the + /// agent_metrics_snapshot 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. + /// + /// 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 growing while + /// / stand + /// still). Momentary queue length is deliberately exported only as a peak. + /// + /// + public sealed class SignalPipelineHealth + { + /// Last SessionSignalOrdinal assigned by the ingress worker; −1 before the first signal. + public long LastAssignedSignalOrdinal { get; set; } + + /// Total signals fully processed (reduce + effects). Monotonic. + public long ProcessedSignalCount { get; set; } + + /// Signals accepted by Post but not yet fully processed (backlog gauge; pairs with the totals). + public long PendingSignalCount { get; set; } + + /// Worker-thread re-entrant posts routed via the worker-local queue. Monotonic. + public long ReentrantPostCount { get; set; } + + /// Highest channel queue length observed at any Post since Start. Monotonic non-decreasing. + public long QueueLengthPeak { get; set; } + + /// Current DecisionState stage name (e.g. "EspDeviceSetup", "WhiteGloveSealed"). Null when no state is available. + public string? DecisionStage { get; set; } + + /// The DecisionState's LastAppliedSignalOrdinal — the reducer-side twin of the ingress counters. Null when no state is available. + public long? LastAppliedSignalOrdinal { get; set; } + + /// Last WhiteGlove-sealing classifier level ("Weak", "Confirmed", …). Null until the classifier produced a verdict. + public string? WhiteGloveSealingLevel { get; set; } + + /// Score of the last WhiteGlove-sealing verdict. Null until the classifier produced a verdict. + public int? WhiteGloveSealingScore { get; set; } + } +}