using System.Text.Json.Nodes; using Nexus.Api.Data; using Nexus.Api.Models; namespace Nexus.Api.Services; /// /// Small capability-gated adapter around the protocol connector. Keeping /// invocation metadata at this seam lets the connector add W3C trace context /// without leaking Gateway protocol details into the run domain service. /// public sealed class OpenClawRunGateway( IGatewayConnector connector, IOpenClawWriteGate writeGate) : IOpenClawRunGateway { public async Task StartAsync( OpenClawRun run, OpenClawInvocationMetadata invocation, CancellationToken cancellationToken = default) { var writeDecision = await writeGate.EvaluateAsync( "chat.send", "operator.write", cancellationToken); if (!writeDecision.Allowed) { return new OpenClawRunGatewayResult( false, false, OpenClawRunStates.Blocked, writeDecision.Recovery is null ? writeDecision.Message : $"{writeDecision.Message} {writeDecision.Recovery}"); } if (connector.ConnectionState != GatewayConnectionState.Connected) { return new OpenClawRunGatewayResult( true, false, OpenClawRunStates.Blocked, "OpenClaw Gateway is not connected. The run remains durable in Nexus and can be retried."); } if (!connector.Supports("chat.send")) { return new OpenClawRunGatewayResult( false, false, OpenClawRunStates.Unsupported, "The connected OpenClaw Gateway does not advertise chat.send."); } try { var response = await connector.InvokeAsync( "chat.send", new { sessionKey = run.SessionKey, agentId = run.AgentId, message = run.Prompt, deliver = false, idempotencyKey = invocation.IdempotencyKey }, cancellationToken: cancellationToken, invocationContext: ToGatewayContext(invocation, includeIdempotencyParameter: true)); var runId = ReadString(response?["runId"]); var gatewayStatus = ReadString(response?["status"])?.ToLowerInvariant(); var state = gatewayStatus switch { "ok" => OpenClawRunStates.Completed, "started" or "in_flight" => OpenClawRunStates.Running, _ => OpenClawRunStates.Running }; return new OpenClawRunGatewayResult( true, true, state, gatewayStatus is null ? "OpenClaw accepted the run." : $"OpenClaw acknowledged the run as '{gatewayStatus}'.", runId, OpenClawPayloadSanitizer.Redact(response)); } catch (OpenClawGatewayRpcException exception) { return FromException(exception); } } public async Task StopAsync( OpenClawRun run, OpenClawInvocationMetadata invocation, CancellationToken cancellationToken = default) { if (connector.ConnectionState != GatewayConnectionState.Connected) { return new OpenClawRunGatewayResult( true, false, OpenClawRunStates.Blocked, "OpenClaw Gateway is not connected; Nexus did not claim the run was stopped."); } if (string.IsNullOrWhiteSpace(run.OpenClawRunId)) { return new OpenClawRunGatewayResult( true, false, OpenClawRunStates.Blocked, "The run has no exact OpenClaw run id. Nexus will not abort every run in the session."); } if (!connector.Supports("chat.abort") && !connector.Supports("sessions.abort")) { return new OpenClawRunGatewayResult( false, false, OpenClawRunStates.Unsupported, "The connected OpenClaw Gateway advertises neither chat.abort nor sessions.abort."); } var abortMethod = connector.Supports("chat.abort") ? "chat.abort" : "sessions.abort"; var writeDecision = await writeGate.EvaluateAsync( abortMethod, "operator.write", cancellationToken); if (!writeDecision.Allowed) { return new OpenClawRunGatewayResult( false, false, OpenClawRunStates.Blocked, writeDecision.Recovery is null ? writeDecision.Message : $"{writeDecision.Message} {writeDecision.Recovery}"); } try { JsonNode? response; if (connector.Supports("chat.abort")) { response = await connector.InvokeAsync( "chat.abort", new { sessionKey = run.SessionKey, agentId = run.AgentId, runId = run.OpenClawRunId }, cancellationToken: cancellationToken, invocationContext: ToGatewayContext(invocation)); } else { response = await connector.InvokeAsync( "sessions.abort", new { key = run.SessionKey, runId = run.OpenClawRunId, clearQueued = false }, cancellationToken: cancellationToken, invocationContext: ToGatewayContext(invocation)); } return new OpenClawRunGatewayResult( true, true, OpenClawRunStates.Stopped, "OpenClaw accepted the exact run abort.", run.OpenClawRunId, OpenClawPayloadSanitizer.Redact(response)); } catch (OpenClawGatewayRpcException exception) { return FromException(exception); } } public async Task GetHistoryAsync( OpenClawRun run, int limit, CancellationToken cancellationToken = default) { if (connector.ConnectionState != GatewayConnectionState.Connected) { return new OpenClawRunGatewayResult( true, false, OpenClawRunStates.Blocked, "OpenClaw Gateway is disconnected; durable Nexus transitions are still available."); } if (!connector.Supports("chat.history")) { return new OpenClawRunGatewayResult( false, false, OpenClawRunStates.Unsupported, "The connected OpenClaw Gateway does not advertise chat.history."); } try { var response = await connector.InvokeAsync( "chat.history", new { sessionKey = run.SessionKey, agentId = run.AgentId, limit = Math.Clamp(limit, 1, 1000), maxChars = 200_000 }, cancellationToken: cancellationToken); return new OpenClawRunGatewayResult( true, true, "available", "OpenClaw returned display-normalized session history.", run.OpenClawRunId, OpenClawPayloadSanitizer.Redact(response)); } catch (OpenClawGatewayRpcException exception) { return FromException(exception); } } private static OpenClawRunGatewayResult FromException(OpenClawGatewayRpcException exception) { var state = exception.Code is "GATEWAY_DISCONNECTED" or "UNAVAILABLE" ? OpenClawRunStates.Blocked : OpenClawRunStates.Failed; return new OpenClawRunGatewayResult( true, false, state, $"OpenClaw rejected the operation ({exception.Code})."); } private static OpenClawInvocationContext ToGatewayContext( OpenClawInvocationMetadata invocation, bool includeIdempotencyParameter = false) => OpenClawInvocationContext.Create( invocation.Actor, invocation.IdempotencyKey, invocation.CorrelationId, invocation.TraceParent, includeIdempotencyParameter); private static string? ReadString(JsonNode? value) { try { return value?.GetValue(); } catch { return value?.ToJsonString(); } } }