nexus/backend/Services/OpenClawRunService.cs

using System.Text.Json.Nodes;
using Nexus.Api.Data;
using Nexus.Api.Models;
using Nexus.Api.Repositories;

namespace Nexus.Api.Services;

public sealed class OpenClawRunService(
    IOpenClawRunRepository runs,
    IOpenClawRunGateway gateway) : IOpenClawRunService
{
    private const string ResumeCapabilityMessage =
        "OpenClaw exposes chat.send/chat.abort/history, but no durable same-run resume RPC. "
        + "Use Retry to create a correlated new run or start a follow-up turn.";

    public async Task<OpenClawRunCollectionDto> GetAsync(
        OpenClawRunQuery query,
        CancellationToken cancellationToken = default)
    {
        var limit = Math.Clamp(query.Limit, 1, 200);
        var items = await runs.GetAsync(query with { Limit = limit + 1 }, cancellationToken);
        var hasMore = items.Count > limit;
        var selected = items.Take(limit).ToList();
        var nextCursor = hasMore && selected.Count > 0
            ? OpenClawRunCursorCodec.Encode(
                selected[^1].CreatedAt,
                selected[^1].Id)
            : null;

        return new OpenClawRunCollectionDto(
            selected.Select(Map).ToList(),
            nextCursor,
            DateTimeOffset.UtcNow);
    }

    public async Task<OpenClawRunDto?> GetByIdAsync(
        Guid id,
        CancellationToken cancellationToken = default)
    {
        var run = await runs.GetByIdAsync(id, cancellationToken: cancellationToken);
        return run is null ? null : Map(run);
    }

    public async Task<OpenClawRunOperationDto> StartAsync(
        StartOpenClawRunRequest request,
        OpenClawInvocationMetadata invocation,
        CancellationToken cancellationToken = default)
    {
        var existing = await runs.GetByStartIdempotencyKeyAsync(
            invocation.IdempotencyKey,
            cancellationToken);
        if (existing is not null)
        {
            var sameCommand = string.Equals(existing.Prompt, request.Prompt.Trim(), StringComparison.Ordinal)
                              && string.Equals(existing.AgentId, request.AgentId.Trim(), StringComparison.Ordinal)
                              && string.Equals(existing.SessionKey, request.SessionKey.Trim(), StringComparison.Ordinal);
            if (sameCommand
                && existing.Status == OpenClawRunStates.Dispatching
                && string.IsNullOrWhiteSpace(existing.OpenClawRunId))
            {
                var recoverable = await runs.GetByIdAsync(
                    existing.Id,
                    tracking: true,
                    cancellationToken: cancellationToken);
                if (recoverable is not null)
                {
                    var recovered = await gateway.StartAsync(recoverable, invocation, cancellationToken);
                    ApplyDispatchResult(recoverable, recovered);
                    await runs.UpdateAsync(
                        recoverable,
                        History(
                            recoverable,
                            "start_recovery",
                            OpenClawRunStates.Dispatching,
                            recoverable.Status,
                            $"Idempotent dispatch recovery: {recovered.Message}",
                            invocation,
                            idempotencyKey: null),
                        cancellationToken);
                    return Operation(
                        recovered.Ok,
                        recovered.State,
                        recovered.Message,
                        recoverable);
                }
            }

            return Operation(
                sameCommand,
                sameCommand ? "idempotent_replay" : "idempotency_conflict",
                sameCommand
                    ? "The original start result was returned; OpenClaw was not invoked again."
                    : "The idempotency key is already bound to a different run command.",
                existing);
        }

        await ValidateCorrelationsAsync(request.TaskId, request.ProjectId, cancellationToken);
        var now = DateTimeOffset.UtcNow;
        var run = new OpenClawRun
        {
            Title = BuildTitle(request.Title, request.Prompt),
            Prompt = request.Prompt.Trim(),
            AgentId = request.AgentId.Trim(),
            SessionKey = request.SessionKey.Trim(),
            TaskId = request.TaskId,
            ProjectId = request.ProjectId,
            Status = OpenClawRunStates.Dispatching,
            StartIdempotencyKey = invocation.IdempotencyKey,
            CorrelationId = invocation.CorrelationId,
            Actor = invocation.Actor,
            TraceParent = invocation.TraceParent,
            CreatedAt = now,
            UpdatedAt = now
        };
        await runs.AddAsync(
            run,
            History(
                run,
                "start_requested",
                OpenClawRunStates.Dispatching,
                OpenClawRunStates.Dispatching,
                "Nexus durably recorded the run before dispatch.",
                invocation),
            cancellationToken);

        var result = await gateway.StartAsync(run, invocation, cancellationToken);
        ApplyDispatchResult(run, result);
        await runs.UpdateAsync(
            run,
            History(
                run,
                "start_result",
                OpenClawRunStates.Dispatching,
                run.Status,
                result.Message,
                invocation,
                idempotencyKey: null),
            cancellationToken);

        return Operation(result.Ok, result.State, result.Message, run);
    }

    public async Task<OpenClawRunOperationDto?> StopAsync(
        Guid id,
        string? reason,
        OpenClawInvocationMetadata invocation,
        CancellationToken cancellationToken = default)
    {
        var run = await runs.GetByIdAsync(id, tracking: true, cancellationToken: cancellationToken);
        if (run is null)
            return null;

        var replay = await runs.GetInvocationAsync(
            id,
            "stop_requested",
            invocation.IdempotencyKey,
            cancellationToken);
        if (replay is not null)
        {
            return Operation(
                true,
                "idempotent_replay",
                "The original stop result was returned; OpenClaw was not invoked again.",
                run);
        }

        if (OpenClawRunStates.IsTerminal(run.Status))
        {
            var message = $"Run is already terminal with state '{run.Status}'.";
            await runs.UpdateAsync(
                run,
                History(
                    run,
                    "stop_requested",
                    run.Status,
                    run.Status,
                    message,
                    invocation),
                cancellationToken);
            return Operation(true, "already_terminal", message, run);
        }

        var previousState = run.Status;
        run.Status = OpenClawRunStates.Stopping;
        await runs.UpdateAsync(
            run,
            History(
                run,
                "stop_requested",
                previousState,
                OpenClawRunStates.Stopping,
                BuildReasonMessage("Stop requested.", reason),
                invocation),
            cancellationToken);

        var result = await gateway.StopAsync(run, invocation, cancellationToken);
        run.Status = result.Ok ? OpenClawRunStates.Stopped : previousState;
        run.LastError = result.Ok ? null : result.Message;
        if (result.Ok)
            run.FinishedAt = DateTimeOffset.UtcNow;

        await runs.UpdateAsync(
            run,
            History(
                run,
                "stop_result",
                OpenClawRunStates.Stopping,
                run.Status,
                result.Message,
                invocation,
                idempotencyKey: null),
            cancellationToken);
        return Operation(result.Ok, result.State, result.Message, run);
    }

    public async Task<OpenClawRunOperationDto?> ResumeAsync(
        Guid id,
        string? reason,
        OpenClawInvocationMetadata invocation,
        CancellationToken cancellationToken = default)
    {
        var run = await runs.GetByIdAsync(id, tracking: true, cancellationToken: cancellationToken);
        if (run is null)
            return null;

        var replay = await runs.GetInvocationAsync(
            id,
            "resume_requested",
            invocation.IdempotencyKey,
            cancellationToken);
        if (replay is not null)
        {
            return Operation(
                false,
                OpenClawRunStates.Unsupported,
                ResumeCapabilityMessage,
                run);
        }

        await runs.UpdateAsync(
            run,
            History(
                run,
                "resume_requested",
                run.Status,
                run.Status,
                BuildReasonMessage(ResumeCapabilityMessage, reason),
                invocation),
            cancellationToken);

        return Operation(
            false,
            OpenClawRunStates.Unsupported,
            ResumeCapabilityMessage,
            run);
    }

    public async Task<OpenClawRunOperationDto?> RetryAsync(
        Guid id,
        string? reason,
        OpenClawInvocationMetadata invocation,
        CancellationToken cancellationToken = default)
    {
        var source = await runs.GetByIdAsync(id, tracking: true, cancellationToken: cancellationToken);
        if (source is null)
            return null;

        var existing = await runs.GetByStartIdempotencyKeyAsync(
            invocation.IdempotencyKey,
            cancellationToken);
        if (existing is not null)
        {
            var sameSource = existing.RetriedFromRunId == id;
            if (sameSource
                && existing.Status == OpenClawRunStates.Dispatching
                && string.IsNullOrWhiteSpace(existing.OpenClawRunId))
            {
                var recoverable = await runs.GetByIdAsync(
                    existing.Id,
                    tracking: true,
                    cancellationToken: cancellationToken);
                if (recoverable is not null)
                {
                    var recovered = await gateway.StartAsync(recoverable, invocation, cancellationToken);
                    ApplyDispatchResult(recoverable, recovered);
                    await runs.UpdateAsync(
                        recoverable,
                        History(
                            recoverable,
                            "start_recovery",
                            OpenClawRunStates.Dispatching,
                            recoverable.Status,
                            $"Idempotent retry dispatch recovery: {recovered.Message}",
                            invocation,
                            idempotencyKey: null),
                        cancellationToken);
                    return Operation(
                        recovered.Ok,
                        recovered.State,
                        recovered.Message,
                        source,
                        recoverable);
                }
            }

            return Operation(
                sameSource,
                sameSource ? "idempotent_replay" : "idempotency_conflict",
                sameSource
                    ? "The original retry result was returned; OpenClaw was not invoked again."
                    : "The idempotency key is already bound to another run.",
                source,
                existing);
        }

        var rejectedReplay = await runs.GetInvocationAsync(
            id,
            "retry_rejected",
            invocation.IdempotencyKey,
            cancellationToken);
        if (rejectedReplay is not null)
        {
            return Operation(
                false,
                "invalid_state",
                "Only a terminal, blocked, or unsupported run can be retried.",
                source);
        }

        if (!OpenClawRunStates.IsTerminal(source.Status))
        {
            const string message = "Only a terminal, blocked, or unsupported run can be retried.";
            await runs.UpdateAsync(
                source,
                History(
                    source,
                    "retry_rejected",
                    source.Status,
                    source.Status,
                    BuildReasonMessage(message, reason),
                    invocation),
                cancellationToken);
            return Operation(false, "invalid_state", message, source);
        }

        var now = DateTimeOffset.UtcNow;
        var retry = new OpenClawRun
        {
            Title = source.Title,
            Prompt = source.Prompt,
            AgentId = source.AgentId,
            SessionKey = source.SessionKey,
            TaskId = source.TaskId,
            ProjectId = source.ProjectId,
            RetriedFromRunId = source.Id,
            Status = OpenClawRunStates.Dispatching,
            StartIdempotencyKey = invocation.IdempotencyKey,
            CorrelationId = invocation.CorrelationId,
            Actor = invocation.Actor,
            TraceParent = invocation.TraceParent,
            CreatedAt = now,
            UpdatedAt = now
        };

        await runs.AddRetryAsync(
            source,
            retry,
            History(
                source,
                "retry_requested",
                source.Status,
                source.Status,
                BuildReasonMessage($"Retry created run {retry.Id}.", reason),
                invocation,
                resultRunId: retry.Id),
            History(
                retry,
                "start_requested",
                OpenClawRunStates.Dispatching,
                OpenClawRunStates.Dispatching,
                $"Retry of Nexus run {source.Id} was durably recorded before dispatch.",
                invocation),
            cancellationToken);

        var result = await gateway.StartAsync(retry, invocation, cancellationToken);
        ApplyDispatchResult(retry, result);
        await runs.UpdateAsync(
            retry,
            History(
                retry,
                "start_result",
                OpenClawRunStates.Dispatching,
                retry.Status,
                result.Message,
                invocation,
                idempotencyKey: null),
            cancellationToken);

        return Operation(result.Ok, result.State, result.Message, source, retry);
    }

    public async Task<OpenClawRunHistoryResponse?> GetHistoryAsync(
        Guid id,
        int gatewayLimit = 200,
        CancellationToken cancellationToken = default)
    {
        var run = await runs.GetByIdAsync(id, cancellationToken: cancellationToken);
        if (run is null)
            return null;

        var transitions = await runs.GetHistoryAsync(id, cancellationToken);
        var gatewayHistory = await gateway.GetHistoryAsync(
            run,
            Math.Clamp(gatewayLimit, 1, 1000),
            cancellationToken);

        return new OpenClawRunHistoryResponse(
            Map(run),
            transitions.Select(Map).ToList(),
            gatewayHistory.State,
            gatewayHistory.Data,
            gatewayHistory.Message,
            DateTimeOffset.UtcNow);
    }

    public async Task ReconcileAsync(
        GatewayEventEnvelope gatewayEvent,
        CancellationToken cancellationToken = default)
    {
        if (!IsRunEvent(gatewayEvent.Event))
            return;

        var runId = ReadString(gatewayEvent.Payload?["runId"]);
        var state = ReadString(gatewayEvent.Payload?["state"])?.ToLowerInvariant();
        if (string.IsNullOrWhiteSpace(runId) || string.IsNullOrWhiteSpace(state))
            return;

        var projectedState = MapGatewayState(state);
        if (projectedState is null)
            return;

        var gatewayEventId = OpenClawEventIdentity.Create(gatewayEvent);
        if (await runs.HasGatewayEventAsync(gatewayEventId, cancellationToken))
            return;

        var run = await runs.GetByOpenClawRunIdAsync(runId, cancellationToken);
        if (run is null)
            return;

        var sequence = ReadLong(gatewayEvent.Payload?["seq"]);
        if (sequence.HasValue
            && run.LastGatewaySequence.HasValue
            && sequence.Value <= run.LastGatewaySequence.Value)
        {
            return;
        }

        var previousStatus = run.Status;
        var sequenceGap = sequence.HasValue
                          && run.LastGatewaySequence.HasValue
                          && sequence.Value > run.LastGatewaySequence.Value + 1;
        if (!OpenClawRunStates.IsTerminal(run.Status) || OpenClawRunStates.IsTerminal(projectedState))
            run.Status = projectedState;
        if (sequence.HasValue)
            run.LastGatewaySequence = sequence;
        run.SequenceGapDetected |= sequenceGap;
        if (run.StartedAt is null && projectedState == OpenClawRunStates.Running)
            run.StartedAt = gatewayEvent.ReceivedAt;
        if (OpenClawRunStates.IsTerminal(projectedState))
            run.FinishedAt = gatewayEvent.ReceivedAt;
        if (projectedState == OpenClawRunStates.Failed)
            run.LastError = ReadString(gatewayEvent.Payload?["errorMessage"]) ?? "OpenClaw run failed.";
        else if (projectedState is OpenClawRunStates.Completed or OpenClawRunStates.Stopped)
            run.LastError = null;

        var shouldAudit = !string.Equals(previousStatus, run.Status, StringComparison.Ordinal)
                          || sequenceGap
                          || OpenClawRunStates.IsTerminal(projectedState);
        OpenClawRunHistory? history = null;
        if (shouldAudit)
        {
            history = new OpenClawRunHistory
            {
                RunId = run.Id,
                Action = "gateway_event",
                FromStatus = previousStatus,
                ToStatus = run.Status,
                Message = sequenceGap
                    ? $"Gateway event '{gatewayEvent.Event}' reported a per-run sequence gap."
                    : $"Gateway event '{gatewayEvent.Event}' projected run state '{state}'.",
                Actor = "openclaw-gateway",
                CorrelationId = run.CorrelationId,
                TraceParent = run.TraceParent,
                GatewayEventId = gatewayEventId,
                GatewaySequence = sequence,
                SequenceGapDetected = sequenceGap,
                OccurredAt = gatewayEvent.ReceivedAt
            };
        }

        await runs.UpdateProjectionAsync(run, history, cancellationToken);
    }

    private async Task ValidateCorrelationsAsync(
        Guid? taskId,
        Guid? projectId,
        CancellationToken cancellationToken)
    {
        if (taskId.HasValue && !await runs.TaskExistsAsync(taskId.Value, cancellationToken))
            throw new OpenClawRunValidationException("taskId", "The correlated Nexus task does not exist.");
        if (projectId.HasValue && !await runs.ProjectExistsAsync(projectId.Value, cancellationToken))
            throw new OpenClawRunValidationException("projectId", "The correlated Nexus project does not exist.");
    }

    private static void ApplyDispatchResult(OpenClawRun run, OpenClawRunGatewayResult result)
    {
        run.Status = result.State;
        run.OpenClawRunId = result.OpenClawRunId ?? run.OpenClawRunId;
        run.LastError = result.Ok ? null : result.Message;
        if (result.Ok && run.StartedAt is null)
            run.StartedAt = DateTimeOffset.UtcNow;
        if (result.Ok && result.State == OpenClawRunStates.Completed)
            run.FinishedAt = DateTimeOffset.UtcNow;
    }

    private static OpenClawRunHistory History(
        OpenClawRun run,
        string action,
        string fromStatus,
        string toStatus,
        string message,
        OpenClawInvocationMetadata invocation,
        string? idempotencyKey = "__invocation__",
        Guid? resultRunId = null)
        => new()
        {
            RunId = run.Id,
            Action = action,
            FromStatus = fromStatus,
            ToStatus = toStatus,
            Message = message,
            Actor = invocation.Actor,
            CorrelationId = invocation.CorrelationId,
            IdempotencyKey = idempotencyKey == "__invocation__"
                ? invocation.IdempotencyKey
                : idempotencyKey,
            TraceParent = invocation.TraceParent,
            ResultRunId = resultRunId
        };

    private static OpenClawRunOperationDto Operation(
        bool ok,
        string state,
        string message,
        OpenClawRun run,
        OpenClawRun? resultRun = null)
        => new(
            ok,
            state,
            message,
            Map(run),
            resultRun is null ? null : Map(resultRun),
            DateTimeOffset.UtcNow,
            new OperationResultDto(
                run.CorrelationId,
                state,
                run.Revision,
                new EntityRefDto("run", run.Id.ToString(), run.Title),
                resultRun is null
                    ? []
                    : [new EntityRefDto(
                        "run",
                        resultRun.Id.ToString(),
                        resultRun.Title)],
                run.TraceParent));

    private static OpenClawRunDto Map(OpenClawRun run)
        => new(
            run.Id,
            run.Title,
            run.Prompt,
            run.AgentId,
            run.SessionKey,
            run.Status,
            run.TaskId,
            run.ProjectId,
            run.OpenClawRunId,
            run.RetriedFromRunId,
            run.CorrelationId,
            run.Actor,
            run.LastError,
            run.LastGatewaySequence,
            run.SequenceGapDetected,
            run.Status is OpenClawRunStates.Dispatching or OpenClawRunStates.Running or OpenClawRunStates.Stopping,
            OpenClawRunStates.IsTerminal(run.Status),
            false,
            ResumeCapabilityMessage,
            run.CreatedAt,
            run.UpdatedAt,
            run.StartedAt,
            run.FinishedAt);

    private static OpenClawRunHistoryDto Map(OpenClawRunHistory item)
        => new(
            item.Id,
            item.RunId,
            item.Action,
            item.FromStatus,
            item.ToStatus,
            item.Message,
            item.Actor,
            item.CorrelationId,
            item.IdempotencyKey,
            item.TraceParent,
            item.GatewayEventId,
            item.GatewaySequence,
            item.SequenceGapDetected,
            item.ResultRunId,
            item.OccurredAt);

    private static string BuildTitle(string? requestedTitle, string prompt)
    {
        var title = string.IsNullOrWhiteSpace(requestedTitle)
            ? prompt.Replace("\r", " ", StringComparison.Ordinal)
                .Replace("\n", " ", StringComparison.Ordinal)
                .Trim()
            : requestedTitle.Trim();
        return title.Length <= 160 ? title : title[..157] + "...";
    }

    private static string BuildReasonMessage(string message, string? reason)
        => string.IsNullOrWhiteSpace(reason)
            ? message
            : $"{message} Reason: {reason.Trim()}";

    private static bool IsRunEvent(string eventName)
        => eventName.Equals("chat", StringComparison.OrdinalIgnoreCase)
           || eventName.Equals("agent", StringComparison.OrdinalIgnoreCase)
           || eventName.Equals("session.operation", StringComparison.OrdinalIgnoreCase);

    private static string? MapGatewayState(string state)
        => state switch
        {
            "status" or "delta" or "started" or "running" => OpenClawRunStates.Running,
            "final" or "completed" or "succeeded" => OpenClawRunStates.Completed,
            "aborted" or "cancelled" => OpenClawRunStates.Stopped,
            "error" or "failed" => OpenClawRunStates.Failed,
            _ => null
        };

    private static string? ReadString(JsonNode? node)
    {
        try
        {
            return node?.GetValue<string>();
        }
        catch
        {
            return null;
        }
    }

    private static long? ReadLong(JsonNode? node)
    {
        try
        {
            if (node is null)
                return null;
            return node.GetValueKind() switch
            {
                System.Text.Json.JsonValueKind.Number => node.GetValue<long>(),
                System.Text.Json.JsonValueKind.String when long.TryParse(node.GetValue<string>(), out var value) => value,
                _ => null
            };
        }
        catch
        {
            return null;
        }
    }
}

public sealed class OpenClawRunValidationException(string field, string message) : Exception(message)
{
    public string Field { get; } = field;
}