From 2e2de8ef69f59b746c69e7f4a3e169b71677f013 Mon Sep 17 00:00:00 2001 From: Caleb Eden <58373773+calebeden@users.noreply.github.com> Date: Mon, 13 Jul 2026 15:51:49 -0700 Subject: [PATCH 1/5] Add gateway lifecycle telemetry Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 23d8c6ac-3d44-4b96-adfe-1e8509594f79 --- docs/TELEMETRY.md | 25 ++ .../GatewayConnectionManager.cs | 307 +++++++++++++++++- src/OpenClaw.Shared/OpenClawGatewayClient.cs | 6 + .../Telemetry/OpenClawTelemetry.cs | 50 +++ src/OpenClaw.Tray.WinUI/App.xaml.cs | 16 +- .../OpenTelemetryEndpointConnection.cs | 148 +++++++++ .../Services/OpenTelemetryLogPolicy.cs | 1 + .../GatewayConnectionManagerTests.cs | 129 ++++++++ .../Telemetry/OpenClawTelemetryTests.cs | 27 ++ .../OpenTelemetryEndpointConnectionTests.cs | 120 +++++++ 10 files changed, 819 insertions(+), 10 deletions(-) diff --git a/docs/TELEMETRY.md b/docs/TELEMETRY.md index dfb59f2dd..67c9bf6d3 100644 --- a/docs/TELEMETRY.md +++ b/docs/TELEMETRY.md @@ -78,6 +78,31 @@ The OpenTelemetry log pipeline should not export: If a new log category should be exported, add it deliberately and review the structured fields it can emit. +## Gateway connection lifecycle + +The tray exports gateway lifecycle diagnostics when an endpoint is configured: + +- operator connect traces: `openclaw.connection.operator.connect` and + `openclaw.connection.operator.reconnect` +- coarse operator phase spans: + `openclaw.connection.operator.prepare`, + `openclaw.connection.operator.transport`, and + `openclaw.connection.operator.handshake` +- metrics: `openclaw.connection.attempts`, + `openclaw.connection.attempt.duration`, and + `openclaw.connection.state.transitions` +- structured state logs in the `OpenClaw.Telemetry.Connection` category + +Lifecycle attributes are limited to role, operation, outcome, coarse error +category, and finite operator/node/overall states. Gateway URLs, IDs, device +IDs, pairing request IDs, credentials, error messages, and diagnostic-ring +text are not exported. + +The phase spans distinguish local credential/client/tunnel preparation, WebSocket +transport establishment, and the gateway challenge/hello handshake. They +intentionally do not trace signing, serialization, response parsing, or token +persistence as separate operations. + ## Endpoint handling The endpoint setting is a collector endpoint, not a credential or request-parameter store. Accept plain `http` and `https` collector URLs with optional path prefixes. Reject URLs with embedded user info, query strings, or fragments. diff --git a/src/OpenClaw.Connection/GatewayConnectionManager.cs b/src/OpenClaw.Connection/GatewayConnectionManager.cs index ea4d5d419..3417b1627 100644 --- a/src/OpenClaw.Connection/GatewayConnectionManager.cs +++ b/src/OpenClaw.Connection/GatewayConnectionManager.cs @@ -1,4 +1,7 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; using OpenClaw.Shared; +using OpenClaw.Shared.Telemetry; namespace OpenClaw.Connection; @@ -9,6 +12,33 @@ namespace OpenClaw.Connection; /// public sealed class GatewayConnectionManager : IGatewayConnectionManager { + internal const string OperatorConnectSpanName = "openclaw.connection.operator.connect"; + internal const string OperatorReconnectSpanName = "openclaw.connection.operator.reconnect"; + internal const string OperatorPrepareSpanName = "openclaw.connection.operator.prepare"; + internal const string OperatorTransportSpanName = "openclaw.connection.operator.transport"; + internal const string OperatorHandshakeSpanName = "openclaw.connection.operator.handshake"; + internal const string AttemptsMetricName = "openclaw.connection.attempts"; + internal const string AttemptDurationMetricName = "openclaw.connection.attempt.duration"; + internal const string StateTransitionsMetricName = "openclaw.connection.state.transitions"; + + private const string RoleTag = "openclaw.connection.role"; + private const string OperationTag = "openclaw.connection.operation"; + private const string StateScopeTag = "openclaw.connection.state.scope"; + private const string StateFromTag = "openclaw.connection.state.from"; + private const string StateToTag = "openclaw.connection.state.to"; + private static readonly Counter ConnectionAttempts = OpenClawTelemetry.CreateCounter( + AttemptsMetricName, + unit: "{attempt}", + description: "Number of OpenClaw gateway connection attempts."); + private static readonly Histogram ConnectionAttemptDuration = OpenClawTelemetry.CreateHistogram( + AttemptDurationMetricName, + unit: "ms", + description: "Duration of OpenClaw gateway connection attempts."); + private static readonly Counter ConnectionStateTransitions = OpenClawTelemetry.CreateCounter( + StateTransitionsMetricName, + unit: "{transition}", + description: "Number of OpenClaw gateway connection state transitions."); + private readonly ConnectionStateMachine _stateMachine = new(); private readonly ConnectionDiagnostics _diagnostics; private readonly ICredentialResolver _credentialResolver; @@ -27,6 +57,7 @@ public sealed class GatewayConnectionManager : IGatewayConnectionManager private readonly object _nodeOperationLock = new(); private readonly object _devicePairReconnectLock = new(); private readonly object _disposeLock = new(); + private readonly object _telemetryLock = new(); private long _generation; private CancellationTokenSource? _operationCts; @@ -50,6 +81,8 @@ public sealed class GatewayConnectionManager : IGatewayConnectionManager private string? _forceBootstrapForGatewayRecordId; private bool _activeConnectUsedBootstrapToken; private bool _postBootstrapOperatorReconnectScheduled; + private TelemetryAttempt? _operatorTelemetryAttempt; + private GatewayConnectionSnapshot _lastTelemetrySnapshot = GatewayConnectionSnapshot.Idle; private const string MissingNodeCredentialMessage = "No node credential available. Re-pair this PC or add a shared/bootstrap gateway token."; @@ -119,7 +152,7 @@ public async Task ConnectAsync(string? gatewayId = null) await _transitionSemaphore.WaitAsync(); try { - await ConnectCoreAsync(gatewayId); + await ConnectCoreAsync(gatewayId, "connect"); } finally { @@ -151,7 +184,7 @@ public async Task ConnectNodeOnlyAsync(string? gatewayId = null) } /// Core connect logic. Caller must hold . - private async Task ConnectCoreAsync(string? gatewayId = null) + private async Task ConnectCoreAsync(string? gatewayId = null, string operation = "connect") { var id = gatewayId ?? _registry.ActiveGatewayId; if (id == null) @@ -181,6 +214,7 @@ private async Task ConnectCoreAsync(string? gatewayId = null) // Dispose old client await DisposeActiveClientAsync(); + StartOperatorTelemetryAttempt(operation, gen); // Update snapshot with gateway info _stateMachine.Current = _stateMachine.Current with @@ -233,6 +267,10 @@ private async Task ConnectCoreAsync(string? gatewayId = null) _stateMachine.TryTransition( ConnectionTrigger.AuthenticationFailed, BuildCredentialFailureMessage("operator", credentialResolution)); + CompleteOperatorTelemetryAttempt( + gen, + "failure", + ConnectionErrorCategory.AuthFailure); EmitStateChanged(); return; } @@ -258,6 +296,10 @@ tunnel.SshPort is < 1 or > 65535 || _logger.Warn("[ConnMgr] SSH tunnel config is incomplete"); _diagnostics.Record("tunnel", "SSH tunnel config is incomplete"); _stateMachine.TryTransition(ConnectionTrigger.AuthenticationFailed, "SSH tunnel config is incomplete"); + CompleteOperatorTelemetryAttempt( + gen, + "failure", + ConnectionErrorCategory.SshTunnelFailure); EmitStateChanged(); return; } @@ -271,6 +313,10 @@ tunnel.SshPort is < 1 or > 65535 || _logger.Error($"[ConnMgr] SSH tunnel start failed: {ex.Message}"); _diagnostics.Record("tunnel", "SSH tunnel start failed", ex.Message); _stateMachine.TryTransition(ConnectionTrigger.WebSocketError, $"SSH tunnel failed: {ex.Message}"); + CompleteOperatorTelemetryAttempt( + gen, + "failure", + ConnectionErrorCategory.SshTunnelFailure); EmitStateChanged(); return; } @@ -301,6 +347,11 @@ tunnel.SshPort is < 1 or > 65535 || if (!IsCurrentGatewayAttempt(gen, subscribedGatewayId)) return; _ = HandleAuthenticationFailedAsync(msg, gen); }; + lifecycle.DataClient.TransportConnected += (s, e) => + { + if (!IsCurrentGatewayAttempt(gen, subscribedGatewayId)) return; + TransitionOperatorTelemetryPhase(gen, OperatorHandshakeSpanName); + }; lifecycle.DataClient.HandshakeSucceeded += (s, e) => { if (!IsCurrentGatewayAttempt(gen, subscribedGatewayId)) return; @@ -336,6 +387,7 @@ tunnel.SshPort is < 1 or > 65535 || // Connect (fire and forget — the event handlers will drive state transitions) var ct = _operationCts!.Token; + TransitionOperatorTelemetryPhase(gen, OperatorTransportSpanName); _ = Task.Run(async () => { try @@ -346,6 +398,10 @@ tunnel.SshPort is < 1 or > 65535 || catch (Exception ex) { _logger.Error($"[ConnMgr] Connect failed: {ex.Message}"); + CompleteOperatorTelemetryAttempt( + gen, + "failure", + ConnectionErrorCategory.InternalError); } }, ct); } @@ -492,6 +548,7 @@ public async Task DisconnectAsync() /// Core disconnect logic. Caller must hold . private async Task DisconnectCoreAsync() { + CancelOperatorTelemetryAttempt("canceled", ConnectionErrorCategory.Cancelled); Interlocked.Increment(ref _generation); var oldCts = Interlocked.Exchange(ref _operationCts, null); oldCts?.Cancel(); @@ -512,7 +569,7 @@ public async Task ReconnectAsync() try { await DisconnectCoreAsync(); - await ConnectCoreAsync(); + await ConnectCoreAsync(operation: "reconnect"); } finally { @@ -805,11 +862,19 @@ private async Task HandleOperatorStatusChangedAsync(ConnectionStatus status, lon // Don't overwrite PairingRequired — gateway closes socket after pairing required if (_stateMachine.Current.OperatorState != RoleConnectionState.PairingRequired) _stateMachine.TryTransition(ConnectionTrigger.WebSocketDisconnected); + CompleteOperatorTelemetryAttempt( + gen, + "failure", + ConnectionErrorCategory.ServerClose); break; case ConnectionStatus.Error: _diagnostics.RecordWebSocketEvent("WebSocket error"); if (_stateMachine.Current.OperatorState != RoleConnectionState.PairingRequired) _stateMachine.TryTransition(ConnectionTrigger.WebSocketError, "Transport error"); + CompleteOperatorTelemetryAttempt( + gen, + "failure", + ConnectionErrorCategory.NetworkUnreachable); break; case ConnectionStatus.Connecting: _diagnostics.RecordWebSocketEvent("WebSocket connecting"); @@ -835,6 +900,10 @@ private async Task HandleAuthenticationFailedAsync(string message, long gen) _diagnostics.Record("error", "Authentication failed", message); _stateMachine.TryTransition(ConnectionTrigger.AuthenticationFailed, message); + CompleteOperatorTelemetryAttempt( + gen, + "failure", + ConnectionErrorCategory.AuthFailure); EmitStateChanged(); } finally @@ -886,6 +955,7 @@ private async Task HandleHandshakeSucceededAsync(long gen) var prev = _stateMachine.Current.OverallState; _diagnostics.Record("state", "Handshake succeeded (hello-ok)"); _stateMachine.TryTransition(ConnectionTrigger.HandshakeSucceeded); + CompleteOperatorTelemetryAttempt(gen, "success"); var nodeModeIntended = SyncNodeIntentFromSettings(); if (_operatorTokenRecoveryAttemptedGatewayId == _activeGatewayRecordId) _operatorTokenRecoveryAttemptedGatewayId = null; @@ -1157,6 +1227,10 @@ private async Task HandlePairingRequiredAsync(string? requestId, long gen) var prev = _stateMachine.Current.OverallState; _diagnostics.Record("pairing", $"Pairing required — waiting for approval (requestId={requestId})"); _stateMachine.TryTransition(ConnectionTrigger.PairingPending); + CompleteOperatorTelemetryAttempt( + gen, + "pairing_required", + ConnectionErrorCategory.PairingPending); // Store requestId in snapshot so setup flows can use it for explicit approval _stateMachine.SetOperatorPairingRequestId(requestId); _diagnostics.RecordStateChange(prev, _stateMachine.Current.OverallState); @@ -2034,12 +2108,228 @@ private static string BuildDeviceAutoApprovalFailureDetail(IReadOnlyList private void EmitStateChanged() { var snapshot = _stateMachine.Current; + RecordTelemetryStateTransitions(snapshot); // Always fire when any part of the snapshot changed — not just OverallState. // Node sub-state changes (e.g. Idle→PairingRequired) may not change OverallState // but the UI still needs to update. StateChanged?.Invoke(this, snapshot); } + private void StartOperatorTelemetryAttempt(string operation, long generation) + { + var tags = new[] + { + OpenClawTelemetryTag.String(RoleTag, "operator"), + OpenClawTelemetryTag.String(OperationTag, operation), + OpenClawTelemetryTag.String(OpenClawTelemetryTagKey.Source, "gateway_connection") + }; + var rootActivity = OpenClawTelemetry.StartDetachedActivity( + operation == "connect" ? OperatorConnectSpanName : OperatorReconnectSpanName, + tags); + var attempt = new TelemetryAttempt( + generation, + operation, + Stopwatch.GetTimestamp(), + rootActivity) + { + PhaseActivity = rootActivity == null + ? null + : OpenClawTelemetry.StartDetachedActivity( + OperatorPrepareSpanName, + rootActivity.Context, + tags) + }; + TelemetryAttempt? superseded; + + lock (_telemetryLock) + { + superseded = _operatorTelemetryAttempt; + _operatorTelemetryAttempt = attempt; + } + + if (superseded != null) + FinishOperatorTelemetryAttempt(superseded, "superseded", null); + OpenClawTelemetry.Add(ConnectionAttempts, tags: tags); + } + + private void TransitionOperatorTelemetryPhase(long generation, string spanName) + { + TelemetryAttempt attempt; + Activity? previousPhase; + ActivityContext parentContext; + string operation; + long phaseGeneration; + + lock (_telemetryLock) + { + if (_operatorTelemetryAttempt is not { } active || + active.Generation != generation || + active.Activity == null) + { + return; + } + + attempt = active; + previousPhase = attempt.PhaseActivity; + attempt.PhaseActivity = null; + phaseGeneration = ++attempt.PhaseGeneration; + parentContext = attempt.Activity.Context; + operation = attempt.Operation; + } + + FinishTelemetryActivity(previousPhase, "success", null); + var nextPhase = OpenClawTelemetry.StartDetachedActivity( + spanName, + parentContext, + [ + OpenClawTelemetryTag.String(RoleTag, "operator"), + OpenClawTelemetryTag.String(OperationTag, operation), + OpenClawTelemetryTag.String(OpenClawTelemetryTagKey.Source, "gateway_connection") + ]); + + var accepted = false; + lock (_telemetryLock) + { + if (ReferenceEquals(_operatorTelemetryAttempt, attempt) && + attempt.PhaseGeneration == phaseGeneration) + { + attempt.PhaseActivity = nextPhase; + accepted = true; + } + } + + if (!accepted) + FinishTelemetryActivity(nextPhase, "superseded", null); + } + + private void CompleteOperatorTelemetryAttempt( + long generation, + string outcome, + ConnectionErrorCategory? errorCategory = null) + { + TelemetryAttempt? attempt; + lock (_telemetryLock) + { + if (_operatorTelemetryAttempt is not { } active || + active.Generation != generation) + return; + + attempt = active; + _operatorTelemetryAttempt = null; + } + + FinishOperatorTelemetryAttempt(attempt, outcome, errorCategory); + } + + private void CancelOperatorTelemetryAttempt( + string outcome, + ConnectionErrorCategory? errorCategory) + { + TelemetryAttempt? attempt; + lock (_telemetryLock) + { + attempt = _operatorTelemetryAttempt; + _operatorTelemetryAttempt = null; + } + + if (attempt != null) + FinishOperatorTelemetryAttempt(attempt, outcome, errorCategory); + } + + private static void FinishOperatorTelemetryAttempt( + TelemetryAttempt attempt, + string outcome, + ConnectionErrorCategory? errorCategory) + { + var tags = new List + { + OpenClawTelemetryTag.String(RoleTag, "operator"), + OpenClawTelemetryTag.String(OperationTag, attempt.Operation), + OpenClawTelemetryTag.String(OpenClawTelemetryTagKey.Outcome, outcome) + }; + if (errorCategory.HasValue) + { + tags.Add(OpenClawTelemetryTag.String( + OpenClawTelemetryTagKey.ErrorCategory, + errorCategory.Value.ToString().ToLowerInvariant())); + } + FinishTelemetryActivity(attempt.PhaseActivity, outcome, errorCategory); + FinishTelemetryActivity(attempt.Activity, outcome, errorCategory, tags); + + OpenClawTelemetry.Record( + ConnectionAttemptDuration, + Stopwatch.GetElapsedTime(attempt.StartTimestamp).TotalMilliseconds, + tags); + } + + private static void FinishTelemetryActivity( + Activity? activity, + string outcome, + ConnectionErrorCategory? errorCategory, + IEnumerable? tags = null) + { + if (activity == null) + return; + + if (tags != null) + { + foreach (var tag in tags) + activity.SetTag(tag.Key, tag.Value); + } + else + { + activity.SetTag(OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), outcome); + if (errorCategory.HasValue) + { + activity.SetTag( + OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), + errorCategory.Value.ToString().ToLowerInvariant()); + } + } + + activity.SetStatus( + outcome is "failure" or "pairing_rejected" + ? ActivityStatusCode.Error + : outcome == "success" + ? ActivityStatusCode.Ok + : ActivityStatusCode.Unset); + activity.Stop(); + activity.Dispose(); + } + + private void RecordTelemetryStateTransitions(GatewayConnectionSnapshot snapshot) + { + GatewayConnectionSnapshot previous; + lock (_telemetryLock) + { + previous = _lastTelemetrySnapshot; + _lastTelemetrySnapshot = snapshot; + } + + RecordTelemetryStateTransition("operator", previous.OperatorState, snapshot.OperatorState); + RecordTelemetryStateTransition("node", previous.NodeState, snapshot.NodeState); + RecordTelemetryStateTransition("overall", previous.OverallState, snapshot.OverallState); + } + + private static void RecordTelemetryStateTransition( + string scope, + TState from, + TState to) + where TState : struct, Enum + { + if (EqualityComparer.Default.Equals(from, to)) + return; + + OpenClawTelemetry.Add( + ConnectionStateTransitions, + tags: + [ + OpenClawTelemetryTag.String(StateScopeTag, scope), + OpenClawTelemetryTag.String(StateFromTag, from.ToString().ToLowerInvariant()), + OpenClawTelemetryTag.String(StateToTag, to.ToString().ToLowerInvariant()) + ]); + } + private async Task DisposeActiveClientAsync() { await _nodeStartSemaphore.WaitAsync(); @@ -2139,6 +2429,7 @@ private async Task DisposeCoreAsync() { if (_disposed) return; _disposed = true; + CancelOperatorTelemetryAttempt("disposed", ConnectionErrorCategory.Disposed); _operationCts?.Cancel(); // Unsubscribe from node events before disposing the semaphore @@ -2190,6 +2481,16 @@ private async Task DisposeCoreAsync() } } + private sealed record TelemetryAttempt( + long Generation, + string Operation, + long StartTimestamp, + Activity? Activity) + { + public Activity? PhaseActivity { get; set; } + public long PhaseGeneration { get; set; } + } + private void ObserveBackgroundFault(Task task, string message) { if (task.IsFaulted) diff --git a/src/OpenClaw.Shared/OpenClawGatewayClient.cs b/src/OpenClaw.Shared/OpenClawGatewayClient.cs index 449e3cd38..de46fc5de 100644 --- a/src/OpenClaw.Shared/OpenClawGatewayClient.cs +++ b/src/OpenClaw.Shared/OpenClawGatewayClient.cs @@ -164,9 +164,13 @@ protected override Task ProcessMessageAsync(string json) protected override Task OnConnectedAsync() { ResetUnsupportedMethodFlags(); + RaiseTransportConnected(); return Task.CompletedTask; } + protected void RaiseTransportConnected() => + TransportConnected?.Invoke(this, EventArgs.Empty); + protected override bool ShouldAutoReconnect() { // PairingRequired must stay visible, but approval only takes effect on a fresh socket. @@ -232,6 +236,8 @@ protected override void OnDisposing() public event EventHandler? DeviceTokenReceived; /// Raised when the hello-ok handshake completes successfully. public event EventHandler? HandshakeSucceeded; + /// Raised after the WebSocket transport connects, before the gateway handshake begins. + public event EventHandler? TransportConnected; /// Raised when the gateway requires pairing approval for this device. public event EventHandler? PairingRequired; /// Raised when v3 signature was rejected and client fell back to v2. diff --git a/src/OpenClaw.Shared/Telemetry/OpenClawTelemetry.cs b/src/OpenClaw.Shared/Telemetry/OpenClawTelemetry.cs index 93238fd63..e1a4344f6 100644 --- a/src/OpenClaw.Shared/Telemetry/OpenClawTelemetry.cs +++ b/src/OpenClaw.Shared/Telemetry/OpenClawTelemetry.cs @@ -40,6 +40,56 @@ public static class OpenClawTelemetry return activity; } + /// + /// Starts a manually-controlled span without leaving it as the ambient activity. + /// + /// + /// Use this for operations that begin in one asynchronous callback and finish in another. + /// The caller owns the returned activity and must stop or dispose it. + /// + public static Activity? StartDetachedActivity( + string spanName, + IEnumerable? tags = null, + System.Diagnostics.ActivityKind kind = System.Diagnostics.ActivityKind.Internal, + OpenClawActivitySourceName source = OpenClawActivitySourceName.OpenClaw) + { + var previous = Activity.Current; + try + { + return StartActivity(spanName, tags, kind, source); + } + finally + { + Activity.Current = previous; + } + } + + /// + /// Starts a manually-controlled child span without leaving it as the ambient activity. + /// + public static Activity? StartDetachedActivity( + string spanName, + ActivityContext parentContext, + IEnumerable? tags = null, + System.Diagnostics.ActivityKind kind = System.Diagnostics.ActivityKind.Internal, + OpenClawActivitySourceName source = OpenClawActivitySourceName.OpenClaw) + { + if (string.IsNullOrWhiteSpace(spanName)) + throw new ArgumentException("Span name cannot be empty.", nameof(spanName)); + + var previous = Activity.Current; + try + { + var activity = source.ToActivitySource().StartActivity(spanName, kind, parentContext); + ApplyTags(activity, tags); + return activity; + } + finally + { + Activity.Current = previous; + } + } + /// /// Runs a synchronous action inside a span and automatically marks success, cancellation, or failure. /// diff --git a/src/OpenClaw.Tray.WinUI/App.xaml.cs b/src/OpenClaw.Tray.WinUI/App.xaml.cs index fe507a905..5084dc421 100644 --- a/src/OpenClaw.Tray.WinUI/App.xaml.cs +++ b/src/OpenClaw.Tray.WinUI/App.xaml.cs @@ -496,7 +496,8 @@ private async Task OnLaunchedAsync(LaunchActivatedEventArgs args) _settings = new SettingsManager(); _previousSettingsSnapshot = _settings.ToSettingsData().ToConnectionSnapshot(); _openTelemetryConnection = new OpenTelemetryEndpointConnection(); - ApplyOpenTelemetryEndpointSettings(); + await _openTelemetryConnection.ApplyAsync( + OpenTelemetryEndpointOptions.FromSettings(_settings)); _chatCoordinator = new OpenClawTray.Chat.OpenClawChatCoordinator( _settings, () => _nodeService, @@ -1934,6 +1935,7 @@ private void RaiseChatProviderChanged() /// private void OnManagerStateChanged(object? sender, GatewayConnectionSnapshot snap) { + _openTelemetryConnection?.SendConnectionState(snap); var mapped = ConnectionStatusPresenter.ToLegacyStatus(snap); var connectedSideEffectsKey = snap.OperatorState == RoleConnectionState.Connected ? $"{snap.GatewayId ?? snap.GatewayUrl ?? "unknown"}|{snap.OperatorDeviceId ?? "unknown"}" @@ -4461,12 +4463,6 @@ private async Task ExitApplicationAsync() _chatCoordinator = null; }); - SafeShutdownStep("OpenTelemetry endpoint", () => - { - _openTelemetryConnection?.Dispose(); - _openTelemetryConnection = null; - }); - // Dispose runtime services var connectionManager = _connectionManager; if (connectionManager != null) @@ -4478,6 +4474,12 @@ await SafeShutdownStepAsync("gateway client", async () => _connectionManager = null; } + SafeShutdownStep("OpenTelemetry endpoint", () => + { + _openTelemetryConnection?.Dispose(); + _openTelemetryConnection = null; + }); + var nodeService = _nodeService; if (nodeService != null) { diff --git a/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryEndpointConnection.cs b/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryEndpointConnection.cs index b4bfd665b..bd47838c9 100644 --- a/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryEndpointConnection.cs +++ b/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryEndpointConnection.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Diagnostics; using System.Diagnostics.Metrics; using Microsoft.Extensions.Logging; @@ -7,6 +8,7 @@ using OpenTelemetry.Metrics; using OpenTelemetry.Resources; using OpenTelemetry.Trace; +using OpenClaw.Connection; using OpenClaw.Shared.Telemetry; namespace OpenClawTray.Services; @@ -21,18 +23,29 @@ internal enum OpenTelemetryEndpointConnectionState internal interface IOpenTelemetryProbeSink : IDisposable { void SendProbe(OpenTelemetryEndpointOptions options); + void SendConnectionState(OpenTelemetryConnectionState state); bool ForceFlush(int timeoutMilliseconds); } +internal sealed record OpenTelemetryConnectionState( + string EventName, + string OverallState, + string OperatorState, + string NodeState); + internal sealed class OpenTelemetryEndpointConnection : IDisposable { private readonly object _gate = new(); private readonly Func _sinkFactory; private readonly Action _logInfo; private readonly Action _logWarn; + private readonly ConcurrentQueue _pendingConnectionStates = new(); private IOpenTelemetryProbeSink? _sink; + private OpenTelemetryConnectionState? _lastConnectionState; private OpenTelemetryEndpointOptions _currentOptions = OpenTelemetryEndpointOptions.Disabled; private long _applyGeneration; + private long _sinkGeneration; + private int _connectionStateDrainScheduled; private volatile bool _disposed; public OpenTelemetryEndpointConnection() @@ -80,6 +93,93 @@ internal void Apply(OpenTelemetryEndpointOptions options) Apply(options, generation: null, forceProbe: false); } + public void SendConnectionState(GatewayConnectionSnapshot snapshot) + { + ArgumentNullException.ThrowIfNull(snapshot); + var state = CreateConnectionState(snapshot); + var sinkGeneration = Volatile.Read(ref _sinkGeneration); + + if (TrySendConnectionState(state)) + return; + + _pendingConnectionStates.Enqueue( + new PendingConnectionState(state, sinkGeneration)); + ScheduleConnectionStateDrain(); + } + + private bool TrySendConnectionState(OpenTelemetryConnectionState state) + { + if (!Monitor.TryEnter(_gate)) + return false; + + try + { + if (_disposed || _sink == null || state == _lastConnectionState) + return true; + + SendConnectionStateCore(state); + return true; + } + finally + { + Monitor.Exit(_gate); + } + } + + private void ScheduleConnectionStateDrain() + { + if (Interlocked.CompareExchange(ref _connectionStateDrainScheduled, 1, 0) != 0) + return; + + ThreadPool.UnsafeQueueUserWorkItem( + static connection => connection.DrainConnectionStates(), + this, + preferLocal: false); + } + + private void DrainConnectionStates() + { + try + { + while (_pendingConnectionStates.TryDequeue(out var pending)) + { + lock (_gate) + { + if (_disposed) + return; + + if (pending.SinkGeneration != _sinkGeneration || + _sink == null || + pending.State == _lastConnectionState) + { + continue; + } + + SendConnectionStateCore(pending.State); + } + } + } + finally + { + Interlocked.Exchange(ref _connectionStateDrainScheduled, 0); + if (!_pendingConnectionStates.IsEmpty) + ScheduleConnectionStateDrain(); + } + } + + private void SendConnectionStateCore(OpenTelemetryConnectionState state) + { + try + { + _sink!.SendConnectionState(state); + _lastConnectionState = state; + } + catch (Exception ex) + { + _logWarn($"OpenTelemetry connection state export failed: {ex.Message}"); + } + } + private void Apply(OpenTelemetryEndpointOptions options, long? generation, bool forceProbe) { lock (_gate) @@ -108,7 +208,9 @@ private void Apply(OpenTelemetryEndpointOptions options, long? generation, bool options == _currentOptions) return; + Interlocked.Increment(ref _sinkGeneration); DisposeSink(); + _lastConnectionState = null; IOpenTelemetryProbeSink? newSink = null; try @@ -155,21 +257,25 @@ public void Dispose() { _disposed = true; Interlocked.Increment(ref _applyGeneration); + Interlocked.Increment(ref _sinkGeneration); lock (_gate) { DisposeSink(); State = OpenTelemetryEndpointConnectionState.Disabled; LastError = null; _currentOptions = OpenTelemetryEndpointOptions.Disabled; + _lastConnectionState = null; } } private void Disable() { + Interlocked.Increment(ref _sinkGeneration); DisposeSink(); State = OpenTelemetryEndpointConnectionState.Disabled; LastError = null; _currentOptions = OpenTelemetryEndpointOptions.Disabled; + _lastConnectionState = null; } private void DisposeSink() @@ -197,6 +303,25 @@ private void DisposeProbeSink(IOpenTelemetryProbeSink? sink, string context) private bool IsStale(long? generation) => _disposed || (generation.HasValue && generation.Value != Volatile.Read(ref _applyGeneration)); + + private static OpenTelemetryConnectionState CreateConnectionState( + GatewayConnectionSnapshot snapshot) => + new( + snapshot.OverallState switch + { + OverallConnectionState.Ready => "ready", + OverallConnectionState.Degraded => "degraded", + OverallConnectionState.PairingRequired => "pairing_required", + OverallConnectionState.Error => "error", + _ => "state_changed" + }, + snapshot.OverallState.ToString().ToLowerInvariant(), + snapshot.OperatorState.ToString().ToLowerInvariant(), + snapshot.NodeState.ToString().ToLowerInvariant()); + + private sealed record PendingConnectionState( + OpenTelemetryConnectionState State, + long SinkGeneration); } internal sealed class OpenTelemetryOtlpProbeSink : IOpenTelemetryProbeSink @@ -211,6 +336,7 @@ internal sealed class OpenTelemetryOtlpProbeSink : IOpenTelemetryProbeSink private const string ExporterProtocolTagKey = "openclaw.exporter.protocol"; private const string SignalTagKey = "openclaw.signal"; private static readonly EventId ExporterProbeLogEvent = new(1000, "OpenTelemetryExporterProbeSent"); + private static readonly EventId ConnectionStateLogEvent = new(1100, "GatewayConnectionStateChanged"); private static readonly Counter ExporterProbeCounter = OpenClawTelemetry.CreateCounter( "openclaw.telemetry.exporter.probes", unit: "{probe}", @@ -227,6 +353,7 @@ internal enum OpenTelemetryOtlpSignal private readonly MeterProvider _meterProvider; private readonly OpenTelemetryLoggerPipeline _loggerPipeline; private readonly ILogger _probeLogger; + private readonly ILogger _connectionLogger; private OpenTelemetryOtlpProbeSink( TracerProvider tracerProvider, @@ -237,6 +364,7 @@ private OpenTelemetryOtlpProbeSink( _meterProvider = meterProvider; _loggerPipeline = loggerPipeline; _probeLogger = loggerPipeline.CreateLogger(OpenTelemetryLogPolicy.TelemetryExporterCategory); + _connectionLogger = loggerPipeline.CreateLogger(OpenTelemetryLogPolicy.ConnectionCategory); } public static IOpenTelemetryProbeSink Create(OpenTelemetryEndpointOptions options) @@ -297,6 +425,26 @@ public void SendProbe(OpenTelemetryEndpointOptions options) static (_, _) => "OpenClaw telemetry exporter probe log sent."); } + public void SendConnectionState(OpenTelemetryConnectionState state) + { + var level = state.EventName is "degraded" or "pairing_required" or "error" + ? LogLevel.Warning + : LogLevel.Information; + KeyValuePair[] attributes = + [ + new("openclaw.connection.event", state.EventName), + new("openclaw.connection.state.overall", state.OverallState), + new("openclaw.connection.state.operator", state.OperatorState), + new("openclaw.connection.state.node", state.NodeState) + ]; + _connectionLogger.Log( + level, + ConnectionStateLogEvent, + attributes, + null, + static (_, _) => "OpenClaw gateway connection state changed."); + } + public bool ForceFlush(int timeoutMilliseconds) { if (timeoutMilliseconds < 0) diff --git a/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryLogPolicy.cs b/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryLogPolicy.cs index 6e3955159..e2152f85c 100644 --- a/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryLogPolicy.cs +++ b/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryLogPolicy.cs @@ -5,6 +5,7 @@ namespace OpenClawTray.Services; internal static class OpenTelemetryLogPolicy { public const string TelemetryExporterCategory = "OpenClaw.Telemetry.Exporter"; + public const string ConnectionCategory = "OpenClaw.Telemetry.Connection"; public static bool ShouldExport(string? category, LogLevel level) => level is >= LogLevel.Information and < LogLevel.None && diff --git a/tests/OpenClaw.Connection.Tests/GatewayConnectionManagerTests.cs b/tests/OpenClaw.Connection.Tests/GatewayConnectionManagerTests.cs index e442d88ab..82b2d74d9 100644 --- a/tests/OpenClaw.Connection.Tests/GatewayConnectionManagerTests.cs +++ b/tests/OpenClaw.Connection.Tests/GatewayConnectionManagerTests.cs @@ -1,4 +1,6 @@ +using System.Diagnostics; using OpenClaw.Shared; +using OpenClaw.Shared.Telemetry; using OpenClaw.Connection; namespace OpenClaw.Connection.Tests; @@ -49,6 +51,7 @@ public async Task ConnectAsync_WithNoCredential_TransitionsToError() { SetupGateway("gw-1", "wss://test"); _resolver.OperatorCredential = null; + using var activities = new ActivityCollector(); GatewayConnectionSnapshot? lastSnap = null; _manager.StateChanged += (_, s) => lastSnap = s; @@ -57,6 +60,17 @@ public async Task ConnectAsync_WithNoCredential_TransitionsToError() Assert.Equal(OverallConnectionState.Error, _manager.CurrentSnapshot.OverallState); Assert.NotNull(lastSnap); + var stopped = activities.GetStopped(); + var root = Assert.Single(stopped, activity => + activity.OperationName == GatewayConnectionManager.OperatorConnectSpanName); + var prepare = Assert.Single(stopped, activity => + activity.OperationName == GatewayConnectionManager.OperatorPrepareSpanName); + Assert.Equal(ActivityStatusCode.Error, root.Status); + Assert.Equal(ActivityStatusCode.Error, prepare.Status); + Assert.Equal("failure", root.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + Assert.Equal( + "authfailure", + root.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); } [Fact] @@ -73,6 +87,39 @@ public async Task ConnectAsync_WithCredential_TransitionsToConnecting() Assert.Equal("test", _manager.CurrentSnapshot.OperatorCredentialSource); } + [Fact] + public async Task ConnectAndReconnect_EmitCompletedOperatorSpans() + { + SetupGateway("gw-1", "wss://test"); + _resolver.OperatorCredential = new GatewayCredential("tok", false, "test"); + using var activities = new ActivityCollector(); + + await _manager.ConnectAsync("gw-1"); + Assert.Single(_factory.CreatedClients).SimulateTransportConnected(); + var connected = WaitForOperatorConnectedAsync(); + _factory.CreatedClients[0].SimulateHandshake(); + await connected; + + await _manager.ReconnectAsync(); + _factory.CreatedClients[1].SimulateTransportConnected(); + connected = WaitForOperatorConnectedAsync(); + _factory.CreatedClients[1].SimulateHandshake(); + await connected; + + var stopped = activities.GetStopped(); + var connectRoot = Assert.Single(stopped, activity => + activity.OperationName == GatewayConnectionManager.OperatorConnectSpanName && + activity.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString() == "success"); + var reconnectRoot = Assert.Single(stopped, activity => + activity.OperationName == GatewayConnectionManager.OperatorReconnectSpanName && + activity.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString() == "success"); + + AssertOperatorPhases(stopped, connectRoot); + AssertOperatorPhases(stopped, reconnectRoot); + Assert.Null(connectRoot.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + Assert.Null(reconnectRoot.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + } + /// /// Regression guard for the post-onboarding "don't cancel an in-flight reconnect" /// path in App.OnboardingCompleted. When the V2 GatewayWelcome wizard saves a new @@ -113,12 +160,21 @@ public async Task DisconnectAsync_TransitionsToIdle() { SetupGateway("gw-1", "wss://test"); _resolver.OperatorCredential = new GatewayCredential("tok", false, "test"); + using var activities = new ActivityCollector(); await _manager.ConnectAsync("gw-1"); await _manager.DisconnectAsync(); Assert.Equal(OverallConnectionState.Idle, _manager.CurrentSnapshot.OverallState); Assert.Null(_manager.OperatorClient); + var root = Assert.Single( + activities.GetStopped(), + activity => activity.OperationName == GatewayConnectionManager.OperatorConnectSpanName); + Assert.Equal(ActivityStatusCode.Unset, root.Status); + Assert.Equal("canceled", root.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + Assert.Equal( + "cancelled", + root.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); } [Fact] @@ -971,6 +1027,22 @@ private void SetupGateway(string id, string url, bool isLocal = false) _registry.SetActive(id); } + private Task WaitForOperatorConnectedAsync() + { + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + EventHandler? handler = null; + handler = (_, snapshot) => + { + if (snapshot.OperatorState != RoleConnectionState.Connected) + return; + + _manager.StateChanged -= handler; + completion.TrySetResult(); + }; + _manager.StateChanged += handler; + return completion.Task.WaitAsync(TimeSpan.FromSeconds(2)); + } + private static async Task InvokeHandshakeSucceededAsync(GatewayConnectionManager manager) { var method = typeof(GatewayConnectionManager).GetMethod( @@ -1578,8 +1650,59 @@ public async Task EnsureNodeConnectedAsync_CallerCancellation_PropagatesOperatio await Assert.ThrowsAnyAsync(() => task); } + private static void AssertOperatorPhases(Activity[] stopped, Activity root) + { + foreach (var phaseName in new[] + { + GatewayConnectionManager.OperatorPrepareSpanName, + GatewayConnectionManager.OperatorTransportSpanName, + GatewayConnectionManager.OperatorHandshakeSpanName + }) + { + var phase = Assert.Single(stopped, activity => + activity.OperationName == phaseName && + activity.TraceId == root.TraceId && + activity.ParentSpanId == root.SpanId); + Assert.Equal( + "success", + phase.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString()); + } + } + // ─── Mocks ─── + private sealed class ActivityCollector : IDisposable + { + private readonly object _gate = new(); + private readonly ActivityListener _listener; + private readonly List _stopped = []; + + public ActivityCollector() + { + _listener = new ActivityListener + { + ShouldListenTo = source => + source.Name == OpenClawActivitySourceName.OpenClaw.ToTelemetryName(), + Sample = (ref ActivityCreationOptions _) => + ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = activity => + { + lock (_gate) + _stopped.Add(activity); + } + }; + ActivitySource.AddActivityListener(_listener); + } + + public Activity[] GetStopped() + { + lock (_gate) + return [.. _stopped]; + } + + public void Dispose() => _listener.Dispose(); + } + private sealed class MockCredentialResolver : ICredentialResolver { public GatewayCredential? OperatorCredential { get; set; } @@ -1648,6 +1771,9 @@ public void SimulateStatusChanged(ConnectionStatus status) => public void SimulateAuthFailed(string msg) => AuthenticationFailed?.Invoke(this, msg); + public void SimulateTransportConnected() => + _client.SimulateTransportConnected(); + public void SimulateHandshake() => _client.SimulateHandshakeSucceeded(); @@ -1665,6 +1791,9 @@ private sealed class MockGatewayClient : OpenClawGatewayClient public MockGatewayClient(string url) : base(url, "mock-token", NullLogger.Instance) { } + public void SimulateTransportConnected() => + RaiseTransportConnected(); + /// Simulate a successful hello-ok handshake for testing. public void SimulateHandshakeSucceeded() { diff --git a/tests/OpenClaw.Shared.Tests/Telemetry/OpenClawTelemetryTests.cs b/tests/OpenClaw.Shared.Tests/Telemetry/OpenClawTelemetryTests.cs index 99adeb6ed..1dc202826 100644 --- a/tests/OpenClaw.Shared.Tests/Telemetry/OpenClawTelemetryTests.cs +++ b/tests/OpenClaw.Shared.Tests/Telemetry/OpenClawTelemetryTests.cs @@ -66,6 +66,33 @@ public void StartActivity_WithListener_AllowsManualMarking() Assert.Contains(activity.Tags, tag => tag.Key == OpenClawTelemetryTagKey.Outcome.ToTelemetryName() && tag.Value == "success"); } + [Fact] + public void StartDetachedActivity_PreservesAmbientActivity() + { + using var collector = ActivityCollector.Listen(OpenClawActivitySourceName.OpenClaw.ToTelemetryName()); + using var parent = new Activity("parent").Start(); + + using var detached = OpenClawTelemetry.StartDetachedActivity("test.detached"); + + Assert.NotNull(detached); + Assert.Same(parent, Activity.Current); + } + + [Fact] + public void StartDetachedActivity_WithExplicitParent_CreatesChildAndPreservesAmbientActivity() + { + using var collector = ActivityCollector.Listen(OpenClawActivitySourceName.OpenClaw.ToTelemetryName()); + using var ambient = new Activity("ambient").Start(); + using var parent = OpenClawTelemetry.StartDetachedActivity("test.parent"); + + using var child = OpenClawTelemetry.StartDetachedActivity("test.child", parent!.Context); + + Assert.NotNull(child); + Assert.Equal(parent.TraceId, child.TraceId); + Assert.Equal(parent.SpanId, child.ParentSpanId); + Assert.Same(ambient, Activity.Current); + } + [Fact] public void Trace_WithListener_RecordsSuccessAndTags() { diff --git a/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs b/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs index 025e3ba1d..c37d4b4b5 100644 --- a/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs +++ b/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.Logging; +using OpenClaw.Connection; using OpenClawTray.Services; using OpenClaw.Shared.Telemetry; @@ -26,6 +27,109 @@ public void Apply_DoesNotCreateSink_WhenEndpointIsEmpty() Assert.False(connection.CurrentOptions.IsEnabled); } + [Fact] + public void SendConnectionState_ForwardsOnlyFiniteStateAndDeduplicates() + { + var sink = new FakeProbeSink(); + using var connection = new OpenTelemetryEndpointConnection( + _ => sink, + _ => { }, + _ => { }); + connection.Apply(OpenTelemetryEndpointOptions.Create( + "http://localhost:4318", + OpenTelemetryEndpointProtocol.HttpProtobuf)); + var snapshot = new GatewayConnectionSnapshot + { + OverallState = OverallConnectionState.Ready, + OperatorState = RoleConnectionState.Connected, + NodeState = RoleConnectionState.Connected, + GatewayId = "sensitive-id", + GatewayUrl = "wss://sensitive-host", + OperatorError = "sensitive-error" + }; + + connection.SendConnectionState(snapshot); + connection.SendConnectionState(snapshot); + + Assert.Equal(1, sink.SendConnectionStateCount); + Assert.Equal( + new OpenTelemetryConnectionState("ready", "ready", "connected", "connected"), + sink.LastConnectionState); + } + + [Fact] + public async Task SendConnectionState_DuringApply_DoesNotBlockAndUsesReplacementSink() + { + var replacementFlushStarted = new ManualResetEventSlim(); + var releaseReplacementFlush = new ManualResetEventSlim(); + var sinks = new List(); + using var connection = new OpenTelemetryEndpointConnection( + _ => + { + var sink = new FakeProbeSink(); + if (sinks.Count == 1) + { + sink.OnForceFlush = () => + { + replacementFlushStarted.Set(); + Assert.True(releaseReplacementFlush.Wait(TimeSpan.FromSeconds(5))); + }; + } + + sinks.Add(sink); + return sink; + }, + _ => { }, + _ => { }); + connection.Apply(OpenTelemetryEndpointOptions.Create( + "http://localhost:4317", + OpenTelemetryEndpointProtocol.Grpc)); + + var applyTask = connection.ApplyAsync(OpenTelemetryEndpointOptions.Create( + "http://localhost:4318", + OpenTelemetryEndpointProtocol.HttpProtobuf)); + Assert.True(replacementFlushStarted.Wait(TimeSpan.FromSeconds(5))); + + var sendTask = Task.Run(() => connection.SendConnectionState(CreateReadySnapshot())); + Assert.Same(sendTask, await Task.WhenAny(sendTask, Task.Delay(TimeSpan.FromSeconds(1)))); + + releaseReplacementFlush.Set(); + await applyTask; + Assert.True(SpinWait.SpinUntil( + () => sinks[1].SendConnectionStateCount == 1, + TimeSpan.FromSeconds(5))); + Assert.Equal( + new OpenTelemetryConnectionState("ready", "ready", "connected", "connected"), + sinks[1].LastConnectionState); + } + + [Fact] + public void Apply_NewOptions_ResetsConnectionStateDeduplication() + { + var sinks = new List(); + using var connection = new OpenTelemetryEndpointConnection( + _ => + { + var sink = new FakeProbeSink(); + sinks.Add(sink); + return sink; + }, + _ => { }, + _ => { }); + + connection.Apply(OpenTelemetryEndpointOptions.Create( + "http://localhost:4317", + OpenTelemetryEndpointProtocol.Grpc)); + connection.SendConnectionState(CreateReadySnapshot()); + connection.Apply(OpenTelemetryEndpointOptions.Create( + "http://localhost:4318", + OpenTelemetryEndpointProtocol.HttpProtobuf)); + connection.SendConnectionState(CreateReadySnapshot()); + + Assert.Equal(1, sinks[0].SendConnectionStateCount); + Assert.Equal(1, sinks[1].SendConnectionStateCount); + } + [Fact] public void Probe_UsesGatewayAlignedTelemetryConstants() { @@ -396,6 +500,14 @@ public void FromSettings_CarriesOnlyEndpointAndProtocol() } } + private static GatewayConnectionSnapshot CreateReadySnapshot() => + new() + { + OverallState = OverallConnectionState.Ready, + OperatorState = RoleConnectionState.Connected, + NodeState = RoleConnectionState.Connected + }; + private sealed class FakeProbeSink : IOpenTelemetryProbeSink { public int SendProbeCount { get; private set; } @@ -406,6 +518,8 @@ private sealed class FakeProbeSink : IOpenTelemetryProbeSink public int DisposeCount { get; private set; } public bool ThrowOnDispose { get; set; } public OpenTelemetryEndpointOptions? LastProbeOptions { get; private set; } + public OpenTelemetryConnectionState? LastConnectionState { get; private set; } + public int SendConnectionStateCount { get; private set; } public void SendProbe(OpenTelemetryEndpointOptions options) { @@ -413,6 +527,12 @@ public void SendProbe(OpenTelemetryEndpointOptions options) LastProbeOptions = options; } + public void SendConnectionState(OpenTelemetryConnectionState state) + { + SendConnectionStateCount++; + LastConnectionState = state; + } + public bool ForceFlush(int timeoutMilliseconds) { ForceFlushCount++; From c3dd64c90a0234dbbecd5b2519fbd790a25002e7 Mon Sep 17 00:00:00 2001 From: Caleb Eden <58373773+calebeden@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:24:15 -0700 Subject: [PATCH 2/5] Preserve telemetry state ordering Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 23d8c6ac-3d44-4b96-adfe-1e8509594f79 --- .../OpenTelemetryEndpointConnection.cs | 51 +++++++++--- .../OpenTelemetryEndpointConnectionTests.cs | 83 ++++++++++++++++++- 2 files changed, 117 insertions(+), 17 deletions(-) diff --git a/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryEndpointConnection.cs b/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryEndpointConnection.cs index bd47838c9..55eade68e 100644 --- a/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryEndpointConnection.cs +++ b/src/OpenClaw.Tray.WinUI/Services/OpenTelemetryEndpointConnection.cs @@ -45,6 +45,8 @@ internal sealed class OpenTelemetryEndpointConnection : IDisposable private OpenTelemetryEndpointOptions _currentOptions = OpenTelemetryEndpointOptions.Disabled; private long _applyGeneration; private long _sinkGeneration; + private long _connectionStateSequence; + private long _lastProcessedConnectionStateSequence; private int _connectionStateDrainScheduled; private volatile bool _disposed; @@ -96,28 +98,40 @@ internal void Apply(OpenTelemetryEndpointOptions options) public void SendConnectionState(GatewayConnectionSnapshot snapshot) { ArgumentNullException.ThrowIfNull(snapshot); - var state = CreateConnectionState(snapshot); - var sinkGeneration = Volatile.Read(ref _sinkGeneration); + var pending = new PendingConnectionState( + CreateConnectionState(snapshot), + Volatile.Read(ref _sinkGeneration), + Interlocked.Increment(ref _connectionStateSequence)); - if (TrySendConnectionState(state)) + if (TrySendConnectionState(pending)) return; - _pendingConnectionStates.Enqueue( - new PendingConnectionState(state, sinkGeneration)); + _pendingConnectionStates.Enqueue(pending); ScheduleConnectionStateDrain(); } - private bool TrySendConnectionState(OpenTelemetryConnectionState state) + private bool TrySendConnectionState(PendingConnectionState pending) { if (!Monitor.TryEnter(_gate)) return false; try { - if (_disposed || _sink == null || state == _lastConnectionState) + if (_disposed || + pending.SinkGeneration != _sinkGeneration || + pending.Sequence <= _lastProcessedConnectionStateSequence || + _sink == null) + { return true; + } - SendConnectionStateCore(state); + if (pending.State == _lastConnectionState) + { + _lastProcessedConnectionStateSequence = pending.Sequence; + return true; + } + + SendConnectionStateCore(pending); return true; } finally @@ -149,13 +163,19 @@ private void DrainConnectionStates() return; if (pending.SinkGeneration != _sinkGeneration || + pending.Sequence <= _lastProcessedConnectionStateSequence || _sink == null || pending.State == _lastConnectionState) { + if (pending.SinkGeneration == _sinkGeneration && + pending.Sequence > _lastProcessedConnectionStateSequence) + { + _lastProcessedConnectionStateSequence = pending.Sequence; + } continue; } - SendConnectionStateCore(pending.State); + SendConnectionStateCore(pending); } } } @@ -167,17 +187,21 @@ private void DrainConnectionStates() } } - private void SendConnectionStateCore(OpenTelemetryConnectionState state) + private void SendConnectionStateCore(PendingConnectionState pending) { try { - _sink!.SendConnectionState(state); - _lastConnectionState = state; + _sink!.SendConnectionState(pending.State); + _lastConnectionState = pending.State; } catch (Exception ex) { _logWarn($"OpenTelemetry connection state export failed: {ex.Message}"); } + finally + { + _lastProcessedConnectionStateSequence = pending.Sequence; + } } private void Apply(OpenTelemetryEndpointOptions options, long? generation, bool forceProbe) @@ -321,7 +345,8 @@ private static OpenTelemetryConnectionState CreateConnectionState( private sealed record PendingConnectionState( OpenTelemetryConnectionState State, - long SinkGeneration); + long SinkGeneration, + long Sequence); } internal sealed class OpenTelemetryOtlpProbeSink : IOpenTelemetryProbeSink diff --git a/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs b/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs index c37d4b4b5..1e1efee1a 100644 --- a/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs +++ b/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs @@ -1,3 +1,4 @@ +using System.Reflection; using Microsoft.Extensions.Logging; using OpenClaw.Connection; using OpenClawTray.Services; @@ -130,6 +131,47 @@ public void Apply_NewOptions_ResetsConnectionStateDeduplication() Assert.Equal(1, sinks[1].SendConnectionStateCount); } + [Fact] + public void SendConnectionState_QueuedOlderState_DoesNotFollowNewerState() + { + var sink = new FakeProbeSink(); + using var connection = new OpenTelemetryEndpointConnection( + _ => sink, + _ => { }, + _ => { }); + connection.Apply(OpenTelemetryEndpointOptions.Create( + "http://localhost:4317", + OpenTelemetryEndpointProtocol.Grpc)); + var gate = typeof(OpenTelemetryEndpointConnection) + .GetField("_gate", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(connection)!; + + Monitor.Enter(gate); + try + { + using var queuedSendCompleted = new ManualResetEventSlim(); + var queuedSendThread = new Thread(() => + { + connection.SendConnectionState(CreateConnectingSnapshot()); + queuedSendCompleted.Set(); + }); + queuedSendThread.Start(); + Assert.True(queuedSendCompleted.Wait(TimeSpan.FromSeconds(5))); + connection.SendConnectionState(CreateReadySnapshot()); + } + finally + { + Monitor.Exit(gate); + } + + Assert.True(SpinWait.SpinUntil( + () => sink.SendConnectionStateCount == 1, + TimeSpan.FromSeconds(5))); + Assert.Equal( + [new OpenTelemetryConnectionState("ready", "ready", "connected", "connected")], + sink.ConnectionStates); + } + [Fact] public void Probe_UsesGatewayAlignedTelemetryConstants() { @@ -508,8 +550,19 @@ private static GatewayConnectionSnapshot CreateReadySnapshot() => NodeState = RoleConnectionState.Connected }; + private static GatewayConnectionSnapshot CreateConnectingSnapshot() => + new() + { + OverallState = OverallConnectionState.Connecting, + OperatorState = RoleConnectionState.Connecting, + NodeState = RoleConnectionState.Idle + }; + private sealed class FakeProbeSink : IOpenTelemetryProbeSink { + private readonly List _connectionStates = []; + private readonly object _connectionStateGate = new(); + public int SendProbeCount { get; private set; } public int ForceFlushCount { get; private set; } public bool ForceFlushResult { get; init; } = true; @@ -518,8 +571,30 @@ private sealed class FakeProbeSink : IOpenTelemetryProbeSink public int DisposeCount { get; private set; } public bool ThrowOnDispose { get; set; } public OpenTelemetryEndpointOptions? LastProbeOptions { get; private set; } - public OpenTelemetryConnectionState? LastConnectionState { get; private set; } - public int SendConnectionStateCount { get; private set; } + public OpenTelemetryConnectionState? LastConnectionState + { + get + { + lock (_connectionStateGate) + return _connectionStates.LastOrDefault(); + } + } + public int SendConnectionStateCount + { + get + { + lock (_connectionStateGate) + return _connectionStates.Count; + } + } + public OpenTelemetryConnectionState[] ConnectionStates + { + get + { + lock (_connectionStateGate) + return [.. _connectionStates]; + } + } public void SendProbe(OpenTelemetryEndpointOptions options) { @@ -529,8 +604,8 @@ public void SendProbe(OpenTelemetryEndpointOptions options) public void SendConnectionState(OpenTelemetryConnectionState state) { - SendConnectionStateCount++; - LastConnectionState = state; + lock (_connectionStateGate) + _connectionStates.Add(state); } public bool ForceFlush(int timeoutMilliseconds) From b458e100a79363697af5c84bae21068b730d43b2 Mon Sep 17 00:00:00 2001 From: Caleb Eden <58373773+calebeden@users.noreply.github.com> Date: Mon, 13 Jul 2026 17:53:41 -0700 Subject: [PATCH 3/5] Preserve ambient trace context Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 23d8c6ac-3d44-4b96-adfe-1e8509594f79 --- .../GatewayConnectionManager.cs | 3 +-- .../Telemetry/OpenClawTelemetry.cs | 24 +++++++++++++++++- .../Telemetry/OpenClawTelemetryTests.cs | 25 +++++++++++++++++++ 3 files changed, 49 insertions(+), 3 deletions(-) diff --git a/src/OpenClaw.Connection/GatewayConnectionManager.cs b/src/OpenClaw.Connection/GatewayConnectionManager.cs index 3417b1627..3a205c8b0 100644 --- a/src/OpenClaw.Connection/GatewayConnectionManager.cs +++ b/src/OpenClaw.Connection/GatewayConnectionManager.cs @@ -2293,8 +2293,7 @@ outcome is "failure" or "pairing_rejected" : outcome == "success" ? ActivityStatusCode.Ok : ActivityStatusCode.Unset); - activity.Stop(); - activity.Dispose(); + OpenClawTelemetry.StopDetachedActivity(activity); } private void RecordTelemetryStateTransitions(GatewayConnectionSnapshot snapshot) diff --git a/src/OpenClaw.Shared/Telemetry/OpenClawTelemetry.cs b/src/OpenClaw.Shared/Telemetry/OpenClawTelemetry.cs index e1a4344f6..d88d376a3 100644 --- a/src/OpenClaw.Shared/Telemetry/OpenClawTelemetry.cs +++ b/src/OpenClaw.Shared/Telemetry/OpenClawTelemetry.cs @@ -45,7 +45,9 @@ public static class OpenClawTelemetry /// /// /// Use this for operations that begin in one asynchronous callback and finish in another. - /// The caller owns the returned activity and must stop or dispose it. + /// The caller owns the returned activity and must finish it with + /// so stopping it cannot replace a newer + /// ambient activity with the context captured when this span started. /// public static Activity? StartDetachedActivity( string spanName, @@ -90,6 +92,26 @@ public static class OpenClawTelemetry } } + /// + /// Stops and disposes a detached activity without changing the caller's ambient activity. + /// + public static void StopDetachedActivity(Activity? activity) + { + if (activity == null) + return; + + var current = Activity.Current; + try + { + activity.Stop(); + activity.Dispose(); + } + finally + { + Activity.Current = current; + } + } + /// /// Runs a synchronous action inside a span and automatically marks success, cancellation, or failure. /// diff --git a/tests/OpenClaw.Shared.Tests/Telemetry/OpenClawTelemetryTests.cs b/tests/OpenClaw.Shared.Tests/Telemetry/OpenClawTelemetryTests.cs index 1dc202826..1113dd347 100644 --- a/tests/OpenClaw.Shared.Tests/Telemetry/OpenClawTelemetryTests.cs +++ b/tests/OpenClaw.Shared.Tests/Telemetry/OpenClawTelemetryTests.cs @@ -93,6 +93,31 @@ public void StartDetachedActivity_WithExplicitParent_CreatesChildAndPreservesAmb Assert.Same(ambient, Activity.Current); } + [Fact] + public void StopDetachedActivity_PreservesNewerAmbientActivity() + { + using var collector = ActivityCollector.Listen(OpenClawActivitySourceName.OpenClaw.ToTelemetryName()); + using var original = new Activity("original").Start(); + var detached = OpenClawTelemetry.StartDetachedActivity("test.detached"); + using var newer = new Activity("newer").Start(); + + OpenClawTelemetry.StopDetachedActivity(detached); + + Assert.NotNull(detached); + Assert.True(detached!.IsStopped); + Assert.Same(newer, Activity.Current); + } + + [Fact] + public void StopDetachedActivity_WithNull_IsSafe() + { + using var ambient = new Activity("ambient").Start(); + + OpenClawTelemetry.StopDetachedActivity(null); + + Assert.Same(ambient, Activity.Current); + } + [Fact] public void Trace_WithListener_RecordsSuccessAndTags() { From 6db98c7a053fec961b9e55c3c4f578b1ff6fa16b Mon Sep 17 00:00:00 2001 From: Caleb Eden <58373773+calebeden@users.noreply.github.com> Date: Tue, 14 Jul 2026 12:02:13 -0700 Subject: [PATCH 4/5] Instrument node connection lifecycle Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 23d8c6ac-3d44-4b96-adfe-1e8509594f79 --- docs/TELEMETRY.md | 45 ++- .../GatewayConnectionManager.cs | 342 +++++++++++++++++- src/OpenClaw.Connection/INodeConnector.cs | 9 + src/OpenClaw.Connection/NodeConnector.cs | 58 ++- src/OpenClaw.Shared/WindowsNodeClient.cs | 24 +- .../GatewayConnectionManagerTests.cs | 338 ++++++++++++++++- .../WindowsNodeClientTests.cs | 39 ++ 7 files changed, 830 insertions(+), 25 deletions(-) diff --git a/docs/TELEMETRY.md b/docs/TELEMETRY.md index 67c9bf6d3..001cb20ff 100644 --- a/docs/TELEMETRY.md +++ b/docs/TELEMETRY.md @@ -84,10 +84,16 @@ The tray exports gateway lifecycle diagnostics when an endpoint is configured: - operator connect traces: `openclaw.connection.operator.connect` and `openclaw.connection.operator.reconnect` +- Windows node connect traces: `openclaw.connection.node.connect` and + `openclaw.connection.node.reconnect` - coarse operator phase spans: `openclaw.connection.operator.prepare`, `openclaw.connection.operator.transport`, and `openclaw.connection.operator.handshake` +- coarse Windows node phase spans: + `openclaw.connection.node.prepare`, + `openclaw.connection.node.transport`, and + `openclaw.connection.node.handshake` - metrics: `openclaw.connection.attempts`, `openclaw.connection.attempt.duration`, and `openclaw.connection.state.transitions` @@ -98,10 +104,41 @@ category, and finite operator/node/overall states. Gateway URLs, IDs, device IDs, pairing request IDs, credentials, error messages, and diagnostic-ring text are not exported. -The phase spans distinguish local credential/client/tunnel preparation, WebSocket -transport establishment, and the gateway challenge/hello handshake. They -intentionally do not trace signing, serialization, response parsing, or token -persistence as separate operations. +The operator phase spans distinguish local credential/client/tunnel preparation, +WebSocket transport establishment, and the gateway challenge/hello handshake. +The Windows node initiates its gateway connection: its prepare span includes +credential resolution, client creation, and synchronous capability registration; +its transport span covers the outbound WebSocket; and its handshake span covers +the gateway's `connect.challenge`, the signed connect request, and `hello-ok`. + +A node attempt succeeds only after `hello-ok` yields connected and paired +readiness. Pending approval completes the attempt as `pairing_required`; human +approval wait time is not included in an open span. If the existing node client +later begins automatic transport recovery, the actual retry is recorded as +`openclaw.connection.node.reconnect` beginning with the transport phase. +Manager-driven starts, including the fresh connection after approval, remain +`openclaw.connection.node.connect`. + +An attempt with outcome `superseded` was replaced by a newer local lifecycle +request before it completed. This is not a gateway or authentication failure. +It exists to make overlapping connection orchestration visible instead of +silently dropping work that had already started. A short `superseded` span +followed by a normal connection span commonly means an automatic or previously +queued start raced with a newer explicit start; the replacement attempt owns the +eventual connection result. + +Pairing and classified gateway failures complete from their specific events +before generic connection status handling. If an active attempt instead ends +with an unclassified `Disconnected` status, telemetry uses `server_close` as a +finite, reasonless fallback because that status carries no close cause. +`Disconnected` covers both orderly remote closes and premature transport loss, +so `server_close` does not prove that the gateway intentionally closed the +connection. Other network failures report `Error` and use +`network_unreachable`; this fallback can therefore be less specific without +changing connection behavior. + +The phase spans intentionally do not trace signing, serialization, response +parsing, capability details, or token persistence as separate operations. ## Endpoint handling diff --git a/src/OpenClaw.Connection/GatewayConnectionManager.cs b/src/OpenClaw.Connection/GatewayConnectionManager.cs index 3a205c8b0..3e7c4c57a 100644 --- a/src/OpenClaw.Connection/GatewayConnectionManager.cs +++ b/src/OpenClaw.Connection/GatewayConnectionManager.cs @@ -17,6 +17,11 @@ public sealed class GatewayConnectionManager : IGatewayConnectionManager internal const string OperatorPrepareSpanName = "openclaw.connection.operator.prepare"; internal const string OperatorTransportSpanName = "openclaw.connection.operator.transport"; internal const string OperatorHandshakeSpanName = "openclaw.connection.operator.handshake"; + internal const string NodeConnectSpanName = "openclaw.connection.node.connect"; + internal const string NodeReconnectSpanName = "openclaw.connection.node.reconnect"; + internal const string NodePrepareSpanName = "openclaw.connection.node.prepare"; + internal const string NodeTransportSpanName = "openclaw.connection.node.transport"; + internal const string NodeHandshakeSpanName = "openclaw.connection.node.handshake"; internal const string AttemptsMetricName = "openclaw.connection.attempts"; internal const string AttemptDurationMetricName = "openclaw.connection.attempt.duration"; internal const string StateTransitionsMetricName = "openclaw.connection.state.transitions"; @@ -82,6 +87,7 @@ public sealed class GatewayConnectionManager : IGatewayConnectionManager private bool _activeConnectUsedBootstrapToken; private bool _postBootstrapOperatorReconnectScheduled; private TelemetryAttempt? _operatorTelemetryAttempt; + private TelemetryAttempt? _nodeTelemetryAttempt; private GatewayConnectionSnapshot _lastTelemetrySnapshot = GatewayConnectionSnapshot.Idle; private const string MissingNodeCredentialMessage = @@ -132,6 +138,11 @@ public GatewayConnectionManager( _nodeConnector.StatusChanged += OnNodeStatusChanged; _nodeConnector.PairingStatusChanged += OnNodePairingStatusChanged; _nodeConnector.DeviceTokenReceived += OnNodeDeviceTokenReceived; + if (_nodeConnector is INodeConnectorTelemetryEvents telemetryEvents) + { + telemetryEvents.TransportConnected += OnNodeTransportConnected; + telemetryEvents.ConnectionFailure += OnNodeConnectionFailure; + } } } @@ -475,6 +486,7 @@ tunnel.SshPort is < 1 or > 65535 || BuildCredentialFailureMessage("node", nodeCredentialResolution), preserveCredentialResolution: true); EmitStateChanged(); + RecordNodePreflightTelemetryFailure(ConnectionErrorCategory.AuthFailure); return null; } @@ -489,6 +501,7 @@ tunnel.SshPort is < 1 or > 65535 || _stateMachine.SetNodeCredentialResolution(nodeCredentialResolution); _stateMachine.BlockNodeStart(NodeTunnelStartFailedMessage, preserveCredentialResolution: true); EmitStateChanged(); + RecordNodePreflightTelemetryFailure(ConnectionErrorCategory.SshTunnelFailure); return null; } @@ -550,6 +563,7 @@ private async Task DisconnectCoreAsync() { CancelOperatorTelemetryAttempt("canceled", ConnectionErrorCategory.Cancelled); Interlocked.Increment(ref _generation); + CancelNodeTelemetryAttempt("canceled", ConnectionErrorCategory.Cancelled); var oldCts = Interlocked.Exchange(ref _operationCts, null); oldCts?.Cancel(); oldCts?.Dispose(); @@ -1398,6 +1412,8 @@ private bool IsCurrentNodeAttempt(long lifecycleGeneration, long nodeGeneration) oldNodeOperationCts?.Cancel(); } + CancelNodeTelemetryAttempt("superseded", null); + if (_nodeConnector != null) { try @@ -1433,6 +1449,12 @@ private bool IsCurrentNodeAttempt(long lifecycleGeneration, long nodeGeneration) nodeGeneration = Interlocked.Increment(ref _nodeConnectionGeneration); _nodeOperationCts = nodeOperationCts; } + + StartNodeTelemetryAttempt( + expectedLifecycleGeneration, + nodeGeneration, + "connect", + NodePrepareSpanName); } } finally @@ -1447,6 +1469,8 @@ await BlockNodeStartAsync( preStartBlockerToken, expectedLifecycleGeneration, expectedNodeGeneration); + CancelNodeTelemetryAttempt("superseded", null); + RecordNodePreflightTelemetryFailure(ConnectionErrorCategory.InternalError); return null; } @@ -1458,6 +1482,10 @@ await BlockNodeStartAsync( } catch (OperationCanceledException) when (nodeOperationToken.IsCancellationRequested) { + CompleteNodeTelemetryAttempt( + nodeGeneration, + "canceled", + ConnectionErrorCategory.Cancelled); return null; } finally @@ -1562,6 +1590,7 @@ private async Task StartNodeConnectionCoreAsync( if (_nodeConnector == null) { await BlockNodeStartAsync(MissingNodeConnectorMessage, cancellationToken, expectedLifecycleGeneration, nodeGeneration); + CompleteNodeTelemetryAttempt(nodeGeneration, "failure", ConnectionErrorCategory.InternalError); return false; } @@ -1573,6 +1602,7 @@ private async Task StartNodeConnectionCoreAsync( if (activeGatewayRecordId == null || activeIdentityPath == null) { await BlockNodeStartAsync(MissingActiveGatewayForNodeMessage, cancellationToken, expectedLifecycleGeneration, nodeGeneration); + CompleteNodeTelemetryAttempt(nodeGeneration, "failure", ConnectionErrorCategory.InternalError); return false; } @@ -1581,6 +1611,7 @@ private async Task StartNodeConnectionCoreAsync( { _logger.Warn("[ConnMgr] Cannot start node — gateway record not found"); await BlockNodeStartAsync(MissingGatewayRecordForNodeMessage, cancellationToken, expectedLifecycleGeneration, nodeGeneration); + CompleteNodeTelemetryAttempt(nodeGeneration, "failure", ConnectionErrorCategory.InternalError); return false; } @@ -1627,6 +1658,7 @@ private async Task StartNodeConnectionCoreAsync( { _transitionSemaphore.Release(); } + CompleteNodeTelemetryAttempt(nodeGeneration, "failure", ConnectionErrorCategory.AuthFailure); return false; } @@ -1665,6 +1697,10 @@ await _nodeConnector.ConnectAsync(nodeConnectUrl, nodeCredential, activeIdentity } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { + CompleteNodeTelemetryAttempt( + nodeGeneration, + "canceled", + ConnectionErrorCategory.Cancelled); return false; } catch (Exception ex) @@ -1682,6 +1718,7 @@ await BlockNodeStartAsync( cancellationToken, expectedLifecycleGeneration, nodeGeneration); + CompleteNodeTelemetryAttempt(nodeGeneration, "failure", ConnectionErrorCategory.NetworkUnreachable); return false; } @@ -1689,12 +1726,38 @@ await BlockNodeStartAsync( Interlocked.Read(ref _nodeConnectionGeneration) == nodeGeneration; } - private void OnNodeStatusChanged(object? sender, ConnectionStatus status) => + private void OnNodeStatusChanged(object? sender, ConnectionStatus status) + { + var lifecycleGeneration = Interlocked.Read(ref _generation); + var nodeGeneration = Interlocked.Read(ref _nodeConnectionGeneration); + ObserveNodeTelemetryStatus(status, lifecycleGeneration, nodeGeneration); AsyncEventHandlerGuard.Run( () => OnNodeStatusChangedAsync(status), _logger, nameof(OnNodeStatusChanged), ex => _diagnostics.Record("node", "Node status handler failed", ex.Message)); + } + + private void OnNodeTransportConnected(object? sender, EventArgs e) + { + var lifecycleGeneration = Interlocked.Read(ref _generation); + var nodeGeneration = Interlocked.Read(ref _nodeConnectionGeneration); + if (IsCurrentNodeAttempt(lifecycleGeneration, nodeGeneration)) + TransitionNodeTelemetryPhase(nodeGeneration, NodeHandshakeSpanName); + } + + private void OnNodeConnectionFailure(object? sender, GatewayErrorKind errorKind) + { + var lifecycleGeneration = Interlocked.Read(ref _generation); + var nodeGeneration = Interlocked.Read(ref _nodeConnectionGeneration); + if (!IsCurrentNodeAttempt(lifecycleGeneration, nodeGeneration)) + return; + + CompleteNodeTelemetryAttempt( + nodeGeneration, + "failure", + MapNodeConnectionErrorCategory(errorKind)); + } private void OnNodeDeviceTokenReceived(object? sender, DeviceTokenReceivedEventArgs e) { @@ -1768,6 +1831,25 @@ private void OnNodePairingStatusChanged(object? sender, PairingStatusEventArgs e { var lifecycleGeneration = Interlocked.Read(ref _generation); var nodeGeneration = Interlocked.Read(ref _nodeConnectionGeneration); + if (e.Status == PairingStatus.Pending) + { + CompleteNodeTelemetryAttempt( + nodeGeneration, + "pairing_required", + ConnectionErrorCategory.PairingPending); + } + else if (e.Status == PairingStatus.Rejected) + { + CompleteNodeTelemetryAttempt( + nodeGeneration, + "pairing_rejected", + ConnectionErrorCategory.PairingRejected); + } + else if (e.Status == PairingStatus.Paired && _nodeConnector?.IsConnected == true) + { + CompleteNodeTelemetryAttempt(nodeGeneration, "success"); + } + AsyncEventHandlerGuard.Run( () => OnNodePairingStatusChangedAsync(e, lifecycleGeneration, nodeGeneration), _logger, @@ -2148,7 +2230,7 @@ private void StartOperatorTelemetryAttempt(string operation, long generation) } if (superseded != null) - FinishOperatorTelemetryAttempt(superseded, "superseded", null); + FinishConnectionTelemetryAttempt(superseded, "operator", "superseded", null); OpenClawTelemetry.Add(ConnectionAttempts, tags: tags); } @@ -2218,7 +2300,7 @@ private void CompleteOperatorTelemetryAttempt( _operatorTelemetryAttempt = null; } - FinishOperatorTelemetryAttempt(attempt, outcome, errorCategory); + FinishConnectionTelemetryAttempt(attempt, "operator", outcome, errorCategory); } private void CancelOperatorTelemetryAttempt( @@ -2233,17 +2315,257 @@ private void CancelOperatorTelemetryAttempt( } if (attempt != null) - FinishOperatorTelemetryAttempt(attempt, outcome, errorCategory); + FinishConnectionTelemetryAttempt(attempt, "operator", outcome, errorCategory); + } + + private void ObserveNodeTelemetryStatus( + ConnectionStatus status, + long lifecycleGeneration, + long nodeGeneration) + { + if (!IsCurrentNodeAttempt(lifecycleGeneration, nodeGeneration)) + return; + + switch (status) + { + case ConnectionStatus.Connecting: + if (!TransitionNodeTelemetryPhase(nodeGeneration, NodeTransportSpanName)) + { + StartNodeTelemetryAttempt( + lifecycleGeneration, + nodeGeneration, + "reconnect", + NodeTransportSpanName); + } + break; + case ConnectionStatus.Connected when _nodeConnector?.PairingStatus == PairingStatus.Paired: + CompleteNodeTelemetryAttempt(nodeGeneration, "success"); + break; + case ConnectionStatus.Disconnected: + // Pairing and classified gateway failures complete through their richer + // events first. Disconnected has no reason payload and covers both orderly + // remote closes and premature transport loss, so server_close is the + // existing finite fallback rather than a claim about the underlying cause. + CompleteNodeTelemetryAttempt( + nodeGeneration, + "failure", + ConnectionErrorCategory.ServerClose); + break; + case ConnectionStatus.Error: + CompleteNodeTelemetryAttempt( + nodeGeneration, + "failure", + ConnectionErrorCategory.NetworkUnreachable); + break; + } + } + + private void StartNodeTelemetryAttempt( + long lifecycleGeneration, + long nodeGeneration, + string operation, + string initialPhaseSpanName) + { + if (!IsCurrentNodeAttempt(lifecycleGeneration, nodeGeneration)) + return; + + var tags = new[] + { + OpenClawTelemetryTag.String(RoleTag, "node"), + OpenClawTelemetryTag.String(OperationTag, operation), + OpenClawTelemetryTag.String(OpenClawTelemetryTagKey.Source, "gateway_connection") + }; + var rootActivity = OpenClawTelemetry.StartDetachedActivity( + operation == "connect" ? NodeConnectSpanName : NodeReconnectSpanName, + tags); + var attempt = new TelemetryAttempt( + nodeGeneration, + operation, + Stopwatch.GetTimestamp(), + rootActivity) + { + PhaseActivity = rootActivity == null + ? null + : OpenClawTelemetry.StartDetachedActivity( + initialPhaseSpanName, + rootActivity.Context, + tags), + PhaseName = initialPhaseSpanName + }; + TelemetryAttempt? superseded = null; + var accepted = false; + + lock (_telemetryLock) + { + if (IsCurrentNodeAttempt(lifecycleGeneration, nodeGeneration)) + { + superseded = _nodeTelemetryAttempt; + _nodeTelemetryAttempt = attempt; + accepted = true; + } + } + + if (!accepted) + { + OpenClawTelemetry.Add(ConnectionAttempts, tags: tags); + FinishConnectionTelemetryAttempt(attempt, "node", "superseded", null); + return; + } + + if (superseded != null) + FinishConnectionTelemetryAttempt(superseded, "node", "superseded", null); + OpenClawTelemetry.Add(ConnectionAttempts, tags: tags); + } + + private static void RecordNodePreflightTelemetryFailure(ConnectionErrorCategory errorCategory) + { + var tags = new[] + { + OpenClawTelemetryTag.String(RoleTag, "node"), + OpenClawTelemetryTag.String(OperationTag, "connect"), + OpenClawTelemetryTag.String(OpenClawTelemetryTagKey.Source, "gateway_connection") + }; + var rootActivity = OpenClawTelemetry.StartDetachedActivity(NodeConnectSpanName, tags); + var attempt = new TelemetryAttempt( + Generation: 0, + Operation: "connect", + StartTimestamp: Stopwatch.GetTimestamp(), + Activity: rootActivity) + { + PhaseActivity = rootActivity == null + ? null + : OpenClawTelemetry.StartDetachedActivity( + NodePrepareSpanName, + rootActivity.Context, + tags) + }; + + OpenClawTelemetry.Add(ConnectionAttempts, tags: tags); + FinishConnectionTelemetryAttempt(attempt, "node", "failure", errorCategory); } - private static void FinishOperatorTelemetryAttempt( + private bool TransitionNodeTelemetryPhase(long nodeGeneration, string spanName) + { + TelemetryAttempt attempt; + Activity? previousPhase; + ActivityContext parentContext; + string operation; + long phaseGeneration; + + lock (_telemetryLock) + { + if (_nodeTelemetryAttempt is not { } active || + active.Generation != nodeGeneration) + { + return false; + } + + if (active.PhaseName == spanName) + return true; + + if (active.Activity == null) + { + active.PhaseName = spanName; + return true; + } + + attempt = active; + previousPhase = attempt.PhaseActivity; + attempt.PhaseActivity = null; + attempt.PhaseName = null; + phaseGeneration = ++attempt.PhaseGeneration; + parentContext = attempt.Activity.Context; + operation = attempt.Operation; + } + + FinishTelemetryActivity(previousPhase, "success", null); + var nextPhase = OpenClawTelemetry.StartDetachedActivity( + spanName, + parentContext, + [ + OpenClawTelemetryTag.String(RoleTag, "node"), + OpenClawTelemetryTag.String(OperationTag, operation), + OpenClawTelemetryTag.String(OpenClawTelemetryTagKey.Source, "gateway_connection") + ]); + + var accepted = false; + lock (_telemetryLock) + { + if (ReferenceEquals(_nodeTelemetryAttempt, attempt) && + attempt.PhaseGeneration == phaseGeneration) + { + attempt.PhaseActivity = nextPhase; + attempt.PhaseName = spanName; + accepted = true; + } + } + + if (!accepted) + FinishTelemetryActivity(nextPhase, "superseded", null); + return true; + } + + private void CompleteNodeTelemetryAttempt( + long nodeGeneration, + string outcome, + ConnectionErrorCategory? errorCategory = null) + { + TelemetryAttempt? attempt; + lock (_telemetryLock) + { + if (_nodeTelemetryAttempt is not { } active || + active.Generation != nodeGeneration) + { + return; + } + + attempt = active; + _nodeTelemetryAttempt = null; + } + + FinishConnectionTelemetryAttempt(attempt, "node", outcome, errorCategory); + } + + private void CancelNodeTelemetryAttempt( + string outcome, + ConnectionErrorCategory? errorCategory) + { + TelemetryAttempt? attempt; + lock (_telemetryLock) + { + attempt = _nodeTelemetryAttempt; + _nodeTelemetryAttempt = null; + } + + if (attempt != null) + FinishConnectionTelemetryAttempt(attempt, "node", outcome, errorCategory); + } + + private static ConnectionErrorCategory MapNodeConnectionErrorCategory(GatewayErrorKind errorKind) => + errorKind switch + { + GatewayErrorKind.Auth or + GatewayErrorKind.TokenDrift or + GatewayErrorKind.ScopeMismatch => ConnectionErrorCategory.AuthFailure, + GatewayErrorKind.PairingRequired => ConnectionErrorCategory.PairingPending, + GatewayErrorKind.PairingRejected => ConnectionErrorCategory.PairingRejected, + GatewayErrorKind.RateLimited => ConnectionErrorCategory.RateLimited, + GatewayErrorKind.Tunnel => ConnectionErrorCategory.SshTunnelFailure, + GatewayErrorKind.Network or + GatewayErrorKind.Tls => ConnectionErrorCategory.NetworkUnreachable, + GatewayErrorKind.Server => ConnectionErrorCategory.ServerClose, + _ => ConnectionErrorCategory.ProtocolMismatch + }; + + private static void FinishConnectionTelemetryAttempt( TelemetryAttempt attempt, + string role, string outcome, ConnectionErrorCategory? errorCategory) { var tags = new List { - OpenClawTelemetryTag.String(RoleTag, "operator"), + OpenClawTelemetryTag.String(RoleTag, role), OpenClawTelemetryTag.String(OperationTag, attempt.Operation), OpenClawTelemetryTag.String(OpenClawTelemetryTagKey.Outcome, outcome) }; @@ -2347,6 +2669,7 @@ private async Task DisposeActiveClientAsync() lock (_nodeOperationLock) Interlocked.Increment(ref _nodeConnectionGeneration); + CancelNodeTelemetryAttempt("canceled", ConnectionErrorCategory.Cancelled); } finally { @@ -2429,6 +2752,7 @@ private async Task DisposeCoreAsync() if (_disposed) return; _disposed = true; CancelOperatorTelemetryAttempt("disposed", ConnectionErrorCategory.Disposed); + CancelNodeTelemetryAttempt("disposed", ConnectionErrorCategory.Disposed); _operationCts?.Cancel(); // Unsubscribe from node events before disposing the semaphore @@ -2438,6 +2762,11 @@ private async Task DisposeCoreAsync() _nodeConnector.StatusChanged -= OnNodeStatusChanged; _nodeConnector.PairingStatusChanged -= OnNodePairingStatusChanged; _nodeConnector.DeviceTokenReceived -= OnNodeDeviceTokenReceived; + if (_nodeConnector is INodeConnectorTelemetryEvents telemetryEvents) + { + telemetryEvents.TransportConnected -= OnNodeTransportConnected; + telemetryEvents.ConnectionFailure -= OnNodeConnectionFailure; + } } // Acquire semaphore briefly to ensure no in-flight reconnect/switch is mid-transition. // Use a short timeout — if something is stuck, proceed with disposal anyway, @@ -2487,6 +2816,7 @@ private sealed record TelemetryAttempt( Activity? Activity) { public Activity? PhaseActivity { get; set; } + public string? PhaseName { get; set; } public long PhaseGeneration { get; set; } } diff --git a/src/OpenClaw.Connection/INodeConnector.cs b/src/OpenClaw.Connection/INodeConnector.cs index e391a19df..397f0f2a5 100644 --- a/src/OpenClaw.Connection/INodeConnector.cs +++ b/src/OpenClaw.Connection/INodeConnector.cs @@ -50,6 +50,15 @@ Task ConnectAsync( Task DisconnectAsync(); } +/// +/// Optional telemetry milestones exposed by production node connectors. +/// +public interface INodeConnectorTelemetryEvents +{ + event EventHandler TransportConnected; + event EventHandler ConnectionFailure; +} + public sealed class NodeClientCreatedEventArgs : EventArgs { public NodeClientCreatedEventArgs(WindowsNodeClient client, string? bearerToken) diff --git a/src/OpenClaw.Connection/NodeConnector.cs b/src/OpenClaw.Connection/NodeConnector.cs index 5f439dd6d..da8e2763d 100644 --- a/src/OpenClaw.Connection/NodeConnector.cs +++ b/src/OpenClaw.Connection/NodeConnector.cs @@ -7,7 +7,7 @@ namespace OpenClaw.Connection; /// Capability setup (canvas, screen capture, etc.) is handled by NodeService, /// which has WinUI dependencies and remains in App.xaml.cs for now. /// -public sealed class NodeConnector : INodeConnector +public sealed class NodeConnector : INodeConnector, INodeConnectorTelemetryEvents { private readonly IOpenClawLogger _logger; private readonly ConnectionDiagnostics? _diagnostics; @@ -21,6 +21,8 @@ public sealed class NodeConnector : INodeConnector public event EventHandler? PairingStatusChanged; public event EventHandler? DeviceTokenReceived; public event EventHandler? ClientCreated; + public event EventHandler? TransportConnected; + public event EventHandler? ConnectionFailure; public NodeConnector(IOpenClawLogger logger, ConnectionDiagnostics? diagnostics = null) { @@ -163,20 +165,15 @@ private async Task ConnectCoreAsync( } client.StatusChanged += (s, e) => - { - if (IsCurrentClient(s, generation)) - StatusChanged?.Invoke(this, e); - }; + ForwardIfCurrent(s, generation, e, StatusChanged); + client.TransportConnected += (s, _) => + ForwardIfCurrent(s, generation, EventArgs.Empty, TransportConnected); + client.ConnectionFailure += (s, e) => + ForwardIfCurrent(s, generation, e, ConnectionFailure); client.PairingStatusChanged += (s, e) => - { - if (IsCurrentClient(s, generation)) - PairingStatusChanged?.Invoke(this, e); - }; + ForwardIfCurrent(s, generation, e, PairingStatusChanged); client.DeviceTokenReceived += (s, e) => - { - if (IsCurrentClient(s, generation)) - DeviceTokenReceived?.Invoke(this, e); - }; + ForwardIfCurrent(s, generation, e, DeviceTokenReceived); try { @@ -221,6 +218,41 @@ private bool IsCurrentClient(object? sender, long generation) } } + // Validation and dispatch stay atomic so a retired client cannot publish after its + // replacement. Subscribers must remain synchronous and must not block on connector + // lifecycle work while this lock is held. + private void ForwardIfCurrent( + object? sender, + long generation, + T args, + EventHandler? handler) + { + lock (_clientLifecycleLock) + { + if (Interlocked.Read(ref _clientGeneration) == generation && + ReferenceEquals(sender, _client)) + { + handler?.Invoke(this, args); + } + } + } + + private void ForwardIfCurrent( + object? sender, + long generation, + EventArgs args, + EventHandler? handler) + { + lock (_clientLifecycleLock) + { + if (Interlocked.Read(ref _clientGeneration) == generation && + ReferenceEquals(sender, _client)) + { + handler?.Invoke(this, args); + } + } + } + private void DisconnectIfCurrent(long generation) { lock (_clientLifecycleLock) diff --git a/src/OpenClaw.Shared/WindowsNodeClient.cs b/src/OpenClaw.Shared/WindowsNodeClient.cs index 8ca7c8125..484dcb523 100644 --- a/src/OpenClaw.Shared/WindowsNodeClient.cs +++ b/src/OpenClaw.Shared/WindowsNodeClient.cs @@ -71,6 +71,10 @@ public class WindowsNodeClient : WebSocketClientBase public event EventHandler? DeviceTokenReceived; /// Raised when the hello-ok handshake completes successfully. public event EventHandler? HandshakeSucceeded; + /// Raised after the WebSocket transport connects, before the gateway challenge arrives. + public event EventHandler? TransportConnected; + /// Raised with a finite classification before a terminal handshake error is published. + public event EventHandler? ConnectionFailure; public new bool IsConnected => _isConnected; public string? NodeId => _nodeId; @@ -108,6 +112,12 @@ public class WindowsNodeClient : WebSocketClientBase protected override int ReceiveBufferSize => 65536; protected override string ClientRole => "node"; + + protected override Task OnConnectedAsync() + { + TransportConnected?.Invoke(this, EventArgs.Empty); + return Task.CompletedTask; + } public WindowsNodeClient(string gatewayUrl, string token, string dataPath, IOpenClawLogger? logger = null, string? bootstrapToken = null) : base(gatewayUrl, ResolveRequiredCredential(token, bootstrapToken, dataPath, logger), logger) @@ -704,7 +714,7 @@ private string BuildNodeConnectMessage(string? nonce, long ts) return (new Dictionary { ["token"] = _gatewayToken }, _gatewayToken); } - private void HandleResponse(JsonElement root) + internal void HandleResponse(JsonElement root) { if (root.TryGetProperty("ok", out var okProp) && okProp.ValueKind == JsonValueKind.False) @@ -884,6 +894,7 @@ private void HandleRequestError(JsonElement root) { _rateLimited = true; _logger.Warn($"[NODE] Terminal auth error; stopping reconnect. Error: {TokenSanitizer.Sanitize(error)}"); + ConnectionFailure?.Invoke(this, ClassifyConnectionFailure(error, errorCode)); RaiseStatusChanged(ConnectionStatus.Error); return; } @@ -900,9 +911,20 @@ private void HandleRequestError(JsonElement root) } _logger.Error($"Node registration failed: {TokenSanitizer.Sanitize(error)} (code: {errorCode})"); + ConnectionFailure?.Invoke(this, ClassifyConnectionFailure(error, errorCode)); RaiseStatusChanged(ConnectionStatus.Error); } + private static GatewayErrorKind ClassifyConnectionFailure(string error, string errorCode) + { + if (error.Contains("too many failed", StringComparison.OrdinalIgnoreCase)) + return GatewayErrorKind.RateLimited; + if (error.Contains("origin not allowed", StringComparison.OrdinalIgnoreCase)) + return GatewayErrorKind.Auth; + + return GatewayErrorClassifier.Classify($"{errorCode} {error}"); + } + private bool PayloadTargetsCurrentDevice(JsonElement payload) { if (TryGetString(payload, "deviceId", out var deviceId) && diff --git a/tests/OpenClaw.Connection.Tests/GatewayConnectionManagerTests.cs b/tests/OpenClaw.Connection.Tests/GatewayConnectionManagerTests.cs index 82b2d74d9..30c3f6522 100644 --- a/tests/OpenClaw.Connection.Tests/GatewayConnectionManagerTests.cs +++ b/tests/OpenClaw.Connection.Tests/GatewayConnectionManagerTests.cs @@ -781,6 +781,7 @@ public async Task HandshakeSucceeded_NodeConnectorThrows_ReportsBlockedNode() SetupGateway("gw-remote", "wss://remote.example", isLocal: false); _resolver.OperatorCredential = new GatewayCredential("op-tok", false, "test"); _resolver.NodeCredential = new GatewayCredential("node-tok", false, "test"); + using var activities = new ActivityCollector(); var nodeConnector = new ScriptedNodeConnector { ConnectAction = (_, _) => throw new InvalidOperationException("connector boom") @@ -807,6 +808,256 @@ public async Task HandshakeSucceeded_NodeConnectorThrows_ReportsBlockedNode() snapshot.NodeState == RoleConnectionState.Error && snapshot.NodeError?.Contains("connector boom", StringComparison.OrdinalIgnoreCase) == true); Assert.NotEqual(RoleConnectionState.Connecting, snapshots.Last().NodeState); + var nodeRoot = Assert.Single( + activities.GetStopped(), + activity => activity.OperationName == GatewayConnectionManager.NodeConnectSpanName); + Assert.Equal("failure", nodeRoot.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + Assert.Equal( + "networkunreachable", + nodeRoot.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + } + + [Fact] + public async Task HandshakeSucceeded_NodePaired_EmitsCompletedNodePhaseTree() + { + SetupGateway("gw-remote", "wss://remote.example", isLocal: false); + _resolver.OperatorCredential = new GatewayCredential("op-tok", false, "test"); + _resolver.NodeCredential = new GatewayCredential("node-tok", false, "test"); + using var activities = new ActivityCollector(); + var nodeConnector = new ScriptedNodeConnector + { + ConnectAction = (node, _) => + { + node.SimulateStatus(ConnectionStatus.Connecting); + node.SimulateTransportConnected(); + node.SimulatePairing(PairingStatus.Paired); + node.SimulateStatus(ConnectionStatus.Connected); + } + }; + using var manager = new GatewayConnectionManager( + _resolver, + _factory, + _registry, + NullLogger.Instance, + nodeConnector: nodeConnector, + isNodeEnabled: () => true); + + await manager.ConnectAsync("gw-remote"); + await InvokeHandshakeSucceededAsync(manager); + + var stopped = activities.GetStopped(); + var root = Assert.Single(stopped, activity => + activity.OperationName == GatewayConnectionManager.NodeConnectSpanName); + Assert.Equal(ActivityStatusCode.Ok, root.Status); + Assert.Equal("success", root.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + Assert.Equal("node", root.GetTagItem("openclaw.connection.role")); + Assert.Null(root.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + AssertNodePhases(stopped, root, includePrepare: true); + } + + [Fact] + public async Task HandshakeSucceeded_NodePairingPending_ClosesAttemptBeforeConnectedStatus() + { + SetupGateway("gw-remote", "wss://remote.example", isLocal: false); + _resolver.OperatorCredential = new GatewayCredential("op-tok", false, "test"); + _resolver.NodeCredential = new GatewayCredential("node-tok", false, "test"); + using var activities = new ActivityCollector(); + var nodeConnector = new ScriptedNodeConnector + { + ConnectAction = (node, _) => + { + node.SimulateStatus(ConnectionStatus.Connecting); + node.SimulateTransportConnected(); + node.SimulatePairing(PairingStatus.Pending); + node.SimulateStatus(ConnectionStatus.Connected); + } + }; + using var manager = new GatewayConnectionManager( + _resolver, + _factory, + _registry, + NullLogger.Instance, + nodeConnector: nodeConnector, + isNodeEnabled: () => true); + + await manager.ConnectAsync("gw-remote"); + await InvokeHandshakeSucceededAsync(manager); + + var stopped = activities.GetStopped(); + var root = Assert.Single(stopped, activity => + activity.OperationName == GatewayConnectionManager.NodeConnectSpanName); + Assert.Equal(ActivityStatusCode.Unset, root.Status); + Assert.Equal( + "pairing_required", + root.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + Assert.Equal( + "pairingpending", + root.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + AssertNodePhases(stopped, root, includePrepare: true, terminalOutcome: "pairing_required"); + } + + [Fact] + public async Task NodeAutomaticRecovery_EmitsReconnectTransportAndHandshakePhases() + { + SetupGateway("gw-remote", "wss://remote.example", isLocal: false); + _resolver.OperatorCredential = new GatewayCredential("op-tok", false, "test"); + _resolver.NodeCredential = new GatewayCredential("node-tok", false, "test"); + using var activities = new ActivityCollector(); + var nodeConnector = new ScriptedNodeConnector + { + ConnectAction = (node, _) => + { + node.SimulateStatus(ConnectionStatus.Connecting); + node.SimulateTransportConnected(); + node.SimulatePairing(PairingStatus.Paired); + node.SimulateStatus(ConnectionStatus.Connected); + } + }; + using var manager = new GatewayConnectionManager( + _resolver, + _factory, + _registry, + NullLogger.Instance, + nodeConnector: nodeConnector, + isNodeEnabled: () => true); + + await manager.ConnectAsync("gw-remote"); + await InvokeHandshakeSucceededAsync(manager); + + nodeConnector.SimulateStatus(ConnectionStatus.Connecting); + nodeConnector.SimulateStatus(ConnectionStatus.Connecting); + nodeConnector.SimulateTransportConnected(); + nodeConnector.SimulatePairing(PairingStatus.Paired); + nodeConnector.SimulateStatus(ConnectionStatus.Connected); + + var stopped = activities.GetStopped(); + var reconnectRoot = Assert.Single(stopped, activity => + activity.OperationName == GatewayConnectionManager.NodeReconnectSpanName); + Assert.Equal("success", reconnectRoot.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + AssertNodePhases(stopped, reconnectRoot, includePrepare: false); + } + + [Fact] + public async Task DisconnectAsync_StaleConnectingDuringRetirement_ClosesReconnectAttempt() + { + SetupGateway("gw-remote", "wss://remote.example", isLocal: false); + _resolver.OperatorCredential = new GatewayCredential("op-tok", false, "test"); + _resolver.NodeCredential = new GatewayCredential("node-tok", false, "test"); + using var activities = new ActivityCollector(); + var nodeConnector = new ScriptedNodeConnector + { + ConnectAction = (node, _) => + { + node.SimulateStatus(ConnectionStatus.Connecting); + node.SimulateTransportConnected(); + node.SimulatePairing(PairingStatus.Paired); + node.SimulateStatus(ConnectionStatus.Connected); + } + }; + using var manager = new GatewayConnectionManager( + _resolver, + _factory, + _registry, + NullLogger.Instance, + nodeConnector: nodeConnector, + isNodeEnabled: () => true); + + await manager.ConnectAsync("gw-remote"); + await InvokeHandshakeSucceededAsync(manager); + nodeConnector.DisconnectAction = node => + node.SimulateStatus(ConnectionStatus.Connecting); + + await manager.DisconnectAsync(); + + var reconnectRoot = Assert.Single( + activities.GetStopped(), + activity => activity.OperationName == GatewayConnectionManager.NodeReconnectSpanName); + Assert.Equal( + "canceled", + reconnectRoot.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + } + + [Fact] + public async Task NodeStart_RetirementFailureAfterConnecting_ClosesReconnectAttempt() + { + SetupGateway("gw-remote", "wss://remote.example", isLocal: false); + _resolver.OperatorCredential = new GatewayCredential("op-tok", false, "test"); + _resolver.NodeCredential = new GatewayCredential("node-tok", false, "test"); + using var activities = new ActivityCollector(); + var nodeConnector = new ScriptedNodeConnector + { + DisconnectAction = node => + node.SimulateStatus(ConnectionStatus.Connecting), + DisconnectException = new InvalidOperationException("retirement failed") + }; + using var manager = new GatewayConnectionManager( + _resolver, + _factory, + _registry, + NullLogger.Instance, + nodeConnector: nodeConnector, + isNodeEnabled: () => true); + + await manager.ConnectAsync("gw-remote"); + await InvokeHandshakeSucceededAsync(manager); + + var stopped = activities.GetStopped(); + var reconnectRoots = stopped + .Where(activity => activity.OperationName == GatewayConnectionManager.NodeReconnectSpanName) + .ToArray(); + Assert.Equal(2, reconnectRoots.Length); + Assert.Single(reconnectRoots, activity => + activity.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString() == "canceled"); + Assert.Single(reconnectRoots, activity => + activity.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString() == "superseded"); + var failedConnect = Assert.Single(stopped, activity => + activity.OperationName == GatewayConnectionManager.NodeConnectSpanName && + activity.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString() == "failure"); + Assert.Equal( + "internalerror", + failedConnect.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + } + + [Theory] + [InlineData(GatewayErrorKind.Auth, "authfailure")] + [InlineData(GatewayErrorKind.RateLimited, "ratelimited")] + [InlineData(GatewayErrorKind.Server, "serverclose")] + [InlineData(GatewayErrorKind.Tunnel, "sshtunnelfailure")] + public async Task NodeClassifiedFailure_UsesSpecificTelemetryCategory( + GatewayErrorKind errorKind, + string expectedCategory) + { + SetupGateway("gw-remote", "wss://remote.example", isLocal: false); + _resolver.OperatorCredential = new GatewayCredential("op-tok", false, "test"); + _resolver.NodeCredential = new GatewayCredential("node-tok", false, "test"); + using var activities = new ActivityCollector(); + var nodeConnector = new ScriptedNodeConnector + { + ConnectAction = (node, _) => + { + node.SimulateStatus(ConnectionStatus.Connecting); + node.SimulateTransportConnected(); + node.SimulateConnectionFailure(errorKind); + node.SimulateStatus(ConnectionStatus.Error); + } + }; + using var manager = new GatewayConnectionManager( + _resolver, + _factory, + _registry, + NullLogger.Instance, + nodeConnector: nodeConnector, + isNodeEnabled: () => true); + + await manager.ConnectAsync("gw-remote"); + await InvokeHandshakeSucceededAsync(manager); + + var root = Assert.Single( + activities.GetStopped(), + activity => activity.OperationName == GatewayConnectionManager.NodeConnectSpanName); + Assert.Equal( + expectedCategory, + root.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); } [Fact] @@ -1220,6 +1471,7 @@ public async Task ConnectNodeOnlyAsync_MissingNodeCredential_ReportsBlockedNode( SetupGateway("gw-1", "wss://test"); _resolver.OperatorCredential = null; _resolver.NodeCredential = null; + using var activities = new ActivityCollector(); var node = new CountingNodeConnector(); using var manager = new GatewayConnectionManager( _resolver, @@ -1237,6 +1489,13 @@ public async Task ConnectNodeOnlyAsync_MissingNodeCredential_ReportsBlockedNode( Assert.Equal(OverallConnectionState.Error, manager.CurrentSnapshot.OverallState); Assert.Contains("No node credential", manager.CurrentSnapshot.NodeError); Assert.Null(manager.CurrentSnapshot.NodeCredentialSource); + var root = Assert.Single( + activities.GetStopped(), + activity => activity.OperationName == GatewayConnectionManager.NodeConnectSpanName); + Assert.Equal("failure", root.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + Assert.Equal( + "authfailure", + root.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); } [Fact] @@ -1321,6 +1580,7 @@ public async Task ConnectNodeOnlyAsync_SameGatewaySupersedesPendingNodeConnect() SetupGateway("gw-1", "wss://test"); _resolver.OperatorCredential = new GatewayCredential("operator-token", false, "test"); _resolver.NodeCredential = new GatewayCredential("node-token", false, "test"); + using var activities = new ActivityCollector(); var node = new SupersedingNodeConnector(); using var manager = new GatewayConnectionManager( _resolver, @@ -1348,6 +1608,16 @@ public async Task ConnectNodeOnlyAsync_SameGatewaySupersedesPendingNodeConnect() Assert.DoesNotContain( manager.Diagnostics.GetAll(), diagnostic => diagnostic.Message == "Node connect failed"); + + await manager.DisconnectAsync(); + var nodeRoots = activities.GetStopped() + .Where(activity => activity.OperationName == GatewayConnectionManager.NodeConnectSpanName) + .ToArray(); + Assert.Equal(2, nodeRoots.Length); + Assert.Single(nodeRoots, activity => + activity.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString() == "superseded"); + Assert.Single(nodeRoots, activity => + activity.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString() == "canceled"); } [Theory] @@ -1580,6 +1850,7 @@ public async Task EnsureNodeConnectedAsync_PairingRejected_Throws() SetupGateway("gw-1", "wss://test"); _resolver.OperatorCredential = new GatewayCredential("op", false, "test"); _resolver.NodeCredential = new GatewayCredential("nd", false, "test"); + using var activities = new ActivityCollector(); var node = new ScriptedNodeConnector { ConnectAction = (s, _) => @@ -1597,6 +1868,17 @@ public async Task EnsureNodeConnectedAsync_PairingRejected_Throws() await Assert.ThrowsAsync( () => manager.EnsureNodeConnectedAsync()); + + var root = Assert.Single( + activities.GetStopped(), + activity => activity.OperationName == GatewayConnectionManager.NodeConnectSpanName); + Assert.Equal(ActivityStatusCode.Error, root.Status); + Assert.Equal( + "pairing_rejected", + root.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + Assert.Equal( + "pairingrejected", + root.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); } [Fact] @@ -1669,6 +1951,47 @@ private static void AssertOperatorPhases(Activity[] stopped, Activity root) } } + private static void AssertNodePhases( + Activity[] stopped, + Activity root, + bool includePrepare, + string terminalOutcome = "success") + { + var phaseNames = includePrepare + ? new[] + { + GatewayConnectionManager.NodePrepareSpanName, + GatewayConnectionManager.NodeTransportSpanName, + GatewayConnectionManager.NodeHandshakeSpanName + } + : + [ + GatewayConnectionManager.NodeTransportSpanName, + GatewayConnectionManager.NodeHandshakeSpanName + ]; + + foreach (var phaseName in phaseNames) + { + var phase = Assert.Single(stopped, activity => + activity.OperationName == phaseName && + activity.TraceId == root.TraceId && + activity.ParentSpanId == root.SpanId); + var expectedOutcome = phaseName == GatewayConnectionManager.NodeHandshakeSpanName + ? terminalOutcome + : "success"; + Assert.Equal( + expectedOutcome, + phase.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())?.ToString()); + } + + if (!includePrepare) + { + Assert.DoesNotContain(stopped, activity => + activity.OperationName == GatewayConnectionManager.NodePrepareSpanName && + activity.TraceId == root.TraceId); + } + } + // ─── Mocks ─── private sealed class ActivityCollector : IDisposable @@ -2290,7 +2613,7 @@ public void Dispose() { } /// Test connector that fires StatusChanged / PairingStatusChanged events synchronously /// so tests can drive the manager's state machine through realistic transitions. /// - private sealed class ScriptedNodeConnector : INodeConnector + private sealed class ScriptedNodeConnector : INodeConnector, INodeConnectorTelemetryEvents { public int ConnectCount { get; private set; } public string? LastGatewayUrl { get; private set; } @@ -2304,10 +2627,14 @@ private sealed class ScriptedNodeConnector : INodeConnector /// gateway URL — use SimulateStatus / SimulatePairing to walk the state machine. /// public Action? ConnectAction { get; set; } + public Action? DisconnectAction { get; set; } + public Exception? DisconnectException { get; set; } public event EventHandler? StatusChanged; public event EventHandler? PairingStatusChanged; public event EventHandler? DeviceTokenReceived; + public event EventHandler? TransportConnected; + public event EventHandler? ConnectionFailure; #pragma warning disable CS0067 // ClientCreated unused in current tests public event EventHandler? ClientCreated; #pragma warning restore CS0067 @@ -2333,6 +2660,9 @@ public Task ConnectAsync( public Task DisconnectAsync() { + DisconnectAction?.Invoke(this); + if (DisconnectException != null) + throw DisconnectException; IsConnected = false; PairingStatus = PairingStatus.Unknown; return Task.CompletedTask; @@ -2350,6 +2680,12 @@ public void SimulatePairing(PairingStatus status, string? requestId = null) PairingStatusChanged?.Invoke(this, new PairingStatusEventArgs(status, deviceId: "scripted-node", requestId: requestId)); } + public void SimulateTransportConnected() => + TransportConnected?.Invoke(this, EventArgs.Empty); + + public void SimulateConnectionFailure(GatewayErrorKind errorKind) => + ConnectionFailure?.Invoke(this, errorKind); + public void SimulateDeviceTokenReceived(string token, string role = "node", string[]? scopes = null) => DeviceTokenReceived?.Invoke(this, new DeviceTokenReceivedEventArgs(token, scopes, role)); diff --git a/tests/OpenClaw.Shared.Tests/WindowsNodeClientTests.cs b/tests/OpenClaw.Shared.Tests/WindowsNodeClientTests.cs index 7b0e1be1d..04227a01c 100644 --- a/tests/OpenClaw.Shared.Tests/WindowsNodeClientTests.cs +++ b/tests/OpenClaw.Shared.Tests/WindowsNodeClientTests.cs @@ -79,6 +79,45 @@ public void Constructor_UsesAppVersionForRegistrationAndConnectMessage() } } + [Theory] + [InlineData("rate limit exceeded", GatewayErrorKind.RateLimited)] + [InlineData("too many failed authentication attempts", GatewayErrorKind.RateLimited)] + [InlineData("device token mismatch", GatewayErrorKind.TokenDrift)] + [InlineData("origin not allowed", GatewayErrorKind.Auth)] + [InlineData("gateway internal error", GatewayErrorKind.Server)] + public void HandleResponse_TerminalError_EmitsFiniteFailureClassification( + string message, + GatewayErrorKind expectedKind) + { + var dataPath = Path.Combine(Path.GetTempPath(), $"openclaw-node-test-{Guid.NewGuid():N}"); + Directory.CreateDirectory(dataPath); + + try + { + using var client = new WindowsNodeClient("ws://localhost:18789", "test-token", dataPath); + GatewayErrorKind? actualKind = null; + client.ConnectionFailure += (_, kind) => actualKind = kind; + using var document = JsonDocument.Parse( + $$""" + { + "type": "res", + "ok": false, + "error": { + "message": "{{message}}", + "code": "TEST_ERROR" + } + } + """); + client.HandleResponse(document.RootElement); + + Assert.Equal(expectedKind, actualKind); + } + finally + { + Directory.Delete(dataPath, true); + } + } + /// /// Regression test: when hello-ok includes auth.deviceToken, PairingStatusChanged must /// fire exactly once — not twice (once from the token block and again from the DeviceToken From 1fea0df92d1515423323a717c4838aa5735cb1c1 Mon Sep 17 00:00:00 2001 From: Caleb Eden <58373773+calebeden@users.noreply.github.com> Date: Tue, 14 Jul 2026 13:13:33 -0700 Subject: [PATCH 5/5] Stabilize telemetry concurrency test Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 23d8c6ac-3d44-4b96-adfe-1e8509594f79 --- .../OpenTelemetryEndpointConnectionTests.cs | 28 +++++++++++++++---- 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs b/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs index 1e1efee1a..004506225 100644 --- a/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs +++ b/tests/OpenClaw.Tray.Tests/OpenTelemetryEndpointConnectionTests.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using System.Reflection; using Microsoft.Extensions.Logging; using OpenClaw.Connection; @@ -61,8 +62,8 @@ public void SendConnectionState_ForwardsOnlyFiniteStateAndDeduplicates() [Fact] public async Task SendConnectionState_DuringApply_DoesNotBlockAndUsesReplacementSink() { - var replacementFlushStarted = new ManualResetEventSlim(); - var releaseReplacementFlush = new ManualResetEventSlim(); + using var replacementFlushStarted = new ManualResetEventSlim(); + using var releaseReplacementFlush = new ManualResetEventSlim(); var sinks = new List(); using var connection = new OpenTelemetryEndpointConnection( _ => @@ -91,14 +92,29 @@ public async Task SendConnectionState_DuringApply_DoesNotBlockAndUsesReplacement OpenTelemetryEndpointProtocol.HttpProtobuf)); Assert.True(replacementFlushStarted.Wait(TimeSpan.FromSeconds(5))); - var sendTask = Task.Run(() => connection.SendConnectionState(CreateReadySnapshot())); - Assert.Same(sendTask, await Task.WhenAny(sendTask, Task.Delay(TimeSpan.FromSeconds(1)))); + var stopwatch = Stopwatch.StartNew(); + try + { + connection.SendConnectionState(CreateReadySnapshot()); + stopwatch.Stop(); + Assert.False( + applyTask.IsCompleted, + "Replacement apply completed before its flush was released."); + } + finally + { + stopwatch.Stop(); + releaseReplacementFlush.Set(); + await applyTask; + } - releaseReplacementFlush.Set(); - await applyTask; + Assert.True( + stopwatch.Elapsed < TimeSpan.FromSeconds(1), + $"Connection state send blocked for {stopwatch.Elapsed}."); Assert.True(SpinWait.SpinUntil( () => sinks[1].SendConnectionStateCount == 1, TimeSpan.FromSeconds(5))); + Assert.Equal(0, sinks[0].SendConnectionStateCount); Assert.Equal( new OpenTelemetryConnectionState("ready", "ready", "connected", "connected"), sinks[1].LastConnectionState);