MAF1/MAF1.Web/Web/ConfigurableWorkflowRunner.cs

using System.Text.Json;
using MAF1.Agents;
using MAF1.Agents.FileCity;
using MAF1.Agents.Weather;
using MAF1.Decisions;
using MAF1.PluginContract;
using MAF1.Plugins;
using Microsoft.Agents.AI;

namespace MAF1.Web;

public sealed class ConfigurableWorkflowRunner(
    AIAgent fileCityAgent,
    AIAgent weatherAgent,
    NodeCatalogService catalog,
    PluginProcessRunner plugins,
    WorkflowRunStore runs,
    int decisionTimeoutSeconds)
{
    private static readonly HashSet<string> ReservedConfigKeys = new(StringComparer.OrdinalIgnoreCase)
    {
        "credentialId",
        "askOnEmptyCities",
        "decisionTimeoutSeconds",
    };

    public async Task<WorkflowRunResult> RunAsync(WorkflowGraph graph, CancellationToken cancellationToken)
    {
        try
        {
            Validate(graph);
            NodeCatalogSnapshot snapshot = catalog.Load();
            List<string> order = TopologicalOrder(graph);
            WorkflowSession session = new()
            {
                RunId = Guid.NewGuid().ToString("n")[..12],
                Graph = graph,
                Catalog = snapshot,
                Order = order,
                Outputs = new Dictionary<string, Dictionary<string, object?>>(StringComparer.OrdinalIgnoreCase),
                Steps = [],
                NextIndex = 0,
            };
            runs.TryAdd(session);
            return await ContinueAsync(session, cancellationToken);
        }
        catch (Exception ex)
        {
            return Fail(ex.Message);
        }
    }

    public WorkflowRunResult? Get(string runId) => runs.GetResult(runId);

    public async Task<WorkflowRunResult> DecideAsync(string runId, DecisionAnswer answer, CancellationToken cancellationToken)
    {
        if (!runs.TryGet(runId, out WorkflowSession? session) || session is null)
        {
            return Fail("找不到这次运行，可能已经结束或服务已重启。");
        }

        return await ResolveAsync(session, answer, cancellationToken, session.Pending?.Id);
    }

    private async Task<WorkflowRunResult> ContinueAsync(WorkflowSession session, CancellationToken cancellationToken)
    {
        try
        {
            for (int i = session.NextIndex; i < session.Order.Count; i++)
            {
                string nodeId = session.Order[i];
                WorkflowNode node = session.Graph.Nodes.First(n => n.Id == nodeId);
                Dictionary<string, object?> inputs = ResolveInputs(session.Graph, node, session.Outputs);
                if (!PassEdgeConditions(session.Graph, node, session.Outputs, out string? skipReason))
                {
                    session.Steps.Add(new NodeRunLog
                    {
                        NodeId = node.Id,
                        Type = node.Type,
                        Title = node.Title,
                        Skipped = true,
                        Message = skipReason,
                        Inputs = inputs,
                    });
                    continue;
                }

                NodeRunLog log = await RunNodeAsync(session.Catalog, node, inputs, cancellationToken);
                session.Steps.Add(log);
                session.Outputs[node.Id] = log.Outputs;
                session.NextIndex = i + 1;

                if (ShouldAskEmptyCities(node, log) && i + 1 < session.Order.Count)
                {
                    return PauseForEmptyCities(session, node, log);
                }
            }

            return Complete(session);
        }
        catch (Exception ex)
        {
            WorkflowRunResult failed = Fail(ex.Message, session.RunId, session.Steps);
            session.Completed = true;
            session.LastResult = failed;
            return failed;
        }
    }

    private WorkflowRunResult PauseForEmptyCities(WorkflowSession session, WorkflowNode node, NodeRunLog log)
    {
        string? reason = ReadString(log.Outputs, "reason");
        int timeout = decisionTimeoutSeconds;
        if (node.Config.TryGetValue("decisionTimeoutSeconds", out string? configured)
            && int.TryParse(configured, out int parsed)
            && parsed > 0)
        {
            timeout = parsed;
        }

        DecisionRequest decision = EmptyCityDecision.Create(node.Id, reason, timeout);
        session.Pending = decision;
        session.TimeoutCts.Cancel();
        session.TimeoutCts.Dispose();
        session.TimeoutCts = new CancellationTokenSource();
        session.LastResult = new WorkflowRunResult
        {
            Ok = true,
            Status = WorkflowRunStatus.NeedsDecision,
            RunId = session.RunId,
            Decision = decision,
            AppliedDecision = session.AppliedDecision,
            Steps = session.Steps,
        };
        StartTimeout(session);
        return session.LastResult;
    }

    private void StartTimeout(WorkflowSession session)
    {
        DecisionRequest? decision = session.Pending;
        if (decision is null)
        {
            return;
        }

        CancellationToken token = session.TimeoutCts.Token;
        _ = Task.Run(async () =>
        {
            try
            {
                await Task.Delay(TimeSpan.FromSeconds(decision.TimeoutSeconds), token);
                await ResolveAsync(
                    session,
                    new DecisionAnswer { OptionId = decision.DefaultOptionId, TimedOut = true },
                    CancellationToken.None,
                    decision.Id);
            }
            catch (OperationCanceledException)
            {
            }
            catch (Exception ex)
            {
                session.Completed = true;
                session.LastResult = Fail($"默认方案自动确认失败：{ex.Message}", session.RunId, session.Steps);
            }
        });
    }

    private async Task<WorkflowRunResult> ResolveAsync(
        WorkflowSession session,
        DecisionAnswer answer,
        CancellationToken cancellationToken,
        string? expectedDecisionId)
    {
        await session.Mutex.WaitAsync(cancellationToken);
        try
        {
            if (session.Completed)
            {
                return session.LastResult;
            }

            if (session.Pending is null
                || (!string.IsNullOrEmpty(expectedDecisionId)
                    && !session.Pending.Id.Equals(expectedDecisionId, StringComparison.OrdinalIgnoreCase)))
            {
                return session.LastResult;
            }

            if (!answer.TimedOut)
            {
                WorkflowRunResult? invalid = ValidateAnswer(session, answer);
                if (invalid is not null)
                {
                    return invalid;
                }
            }

            session.TimeoutCts.Cancel();
            ApplyAnswer(session, answer);
            session.Pending = null;
            return await ContinueAsync(session, cancellationToken);
        }
        catch (Exception ex)
        {
            return Fail(ex.Message, session.RunId, session.Steps);
        }
        finally
        {
            session.Mutex.Release();
        }
    }

    private static WorkflowRunResult? ValidateAnswer(WorkflowSession session, DecisionAnswer answer)
    {
        DecisionRequest decision = session.Pending!;
        DecisionOption? option = decision.Options.FirstOrDefault(
            item => item.Id.Equals(answer.OptionId, StringComparison.OrdinalIgnoreCase));
        if (option is null)
        {
            return InvalidDecision(session, "请选择一个列出的方案。");
        }

        if (option.RequiresText && EmptyCityDecision.ParseCities(answer.Text).Count == 0)
        {
            return InvalidDecision(session, "这个方案需要输入城市名，例如：成都、北京。");
        }

        return null;
    }

    private static WorkflowRunResult InvalidDecision(WorkflowSession session, string error)
        => new()
        {
            Ok = false,
            Status = WorkflowRunStatus.NeedsDecision,
            RunId = session.RunId,
            Error = error,
            Decision = session.Pending,
            AppliedDecision = session.AppliedDecision,
            Steps = session.Steps,
        };

    private static void ApplyAnswer(WorkflowSession session, DecisionAnswer answer)
    {
        DecisionRequest decision = session.Pending
            ?? throw new InvalidOperationException("没有待确认的方案。");
        string optionId = string.IsNullOrWhiteSpace(answer.OptionId)
            ? decision.DefaultOptionId
            : answer.OptionId;
        if (answer.TimedOut)
        {
            optionId = decision.DefaultOptionId;
        }

        DecisionOption option = decision.Options.First(
            item => item.Id.Equals(optionId, StringComparison.OrdinalIgnoreCase));
        DecisionAnswer applied = new()
        {
            OptionId = option.Id,
            Text = answer.Text,
            TimedOut = answer.TimedOut,
        };
        session.AppliedDecision = applied;

        NodeRunLog? step = session.Steps.LastOrDefault(item => item.NodeId == decision.NodeId);
        if (option.Id == DecisionOptionIds.QueryCities)
        {
            List<string> cities = EmptyCityDecision.ParseCities(answer.Text);
            if (!session.Outputs.TryGetValue(decision.NodeId, out Dictionary<string, object?>? outputs))
            {
                outputs = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
                session.Outputs[decision.NodeId] = outputs;
            }

            outputs["hasValidCities"] = true;
            outputs["cities"] = cities;
            outputs["reason"] = $"用户指定城市：{string.Join("、", cities)}";
            if (step is not null)
            {
                step.Outputs = outputs;
                step.Message = Append(step.Message, $"用户确认：改查 {string.Join("、", cities)}。");
            }
        }
        else if (step is not null)
        {
            string how = answer.TimedOut ? "超时，自动采用默认方案" : "用户确认默认方案";
            step.Message = Append(step.Message, $"{how}：结束后续查询。");
        }
    }

    private static string Append(string? message, string extra)
        => string.IsNullOrWhiteSpace(message) ? extra : $"{message}\n{extra}";

    private static WorkflowRunResult Complete(WorkflowSession session)
    {
        WorkflowRunResult result = new()
        {
            Ok = true,
            Status = WorkflowRunStatus.Completed,
            RunId = session.RunId,
            AppliedDecision = session.AppliedDecision,
            Steps = session.Steps,
        };
        session.Completed = true;
        session.Pending = null;
        session.LastResult = result;
        return result;
    }

    private static WorkflowRunResult Fail(string error, string? runId = null, List<NodeRunLog>? steps = null)
        => new()
        {
            Ok = false,
            Status = WorkflowRunStatus.Failed,
            RunId = runId,
            Error = error,
            Steps = steps ?? [],
        };

    private static bool ShouldAskEmptyCities(WorkflowNode node, NodeRunLog log)
    {
        if (log.Skipped)
        {
            return false;
        }

        if (node.Config.TryGetValue("askOnEmptyCities", out string? flag)
            && flag.Equals("false", StringComparison.OrdinalIgnoreCase))
        {
            return false;
        }

        if (!log.Outputs.ContainsKey("hasValidCities"))
        {
            return false;
        }

        return !ReadBool(log.Outputs, "hasValidCities");
    }

    private async Task<NodeRunLog> RunNodeAsync(
        NodeCatalogSnapshot snapshot,
        WorkflowNode node,
        Dictionary<string, object?> inputs,
        CancellationToken cancellationToken)
    {
        LoadedPlugin? plugin = snapshot.LoadedPlugins
            .FirstOrDefault(p => p.Manifest.Id.Equals(node.Type, StringComparison.OrdinalIgnoreCase));
        if (plugin is not null)
        {
            return await RunPluginAsync(plugin, node, inputs, cancellationToken);
        }

        AgentTypeInfo? system = AgentCatalog.FindSystem(node.Type)
            ?? snapshot.System.FirstOrDefault(item => item.Type.Equals(node.Type, StringComparison.OrdinalIgnoreCase));
        if (system is not null)
        {
            AgentStepResult step = system.Handler switch
            {
                SystemHandler.FileCity => await FileCityAgent.RunAsync(fileCityAgent, inputs, cancellationToken),
                SystemHandler.Weather => await WeatherAgent.RunAsync(weatherAgent, inputs, cancellationToken),
                _ => throw new InvalidOperationException(
                    $"系统节点 `{system.Type}` 在 AgentCatalog 里没有绑定实现。请给它设置 Handler。"),
            };
            return ToLog(node, step);
        }

        throw new InvalidOperationException(
            $"未找到节点类型 `{node.Type}`。系统节点来自 AgentCatalog，插件节点来自 plugins 目录，请点「刷新节点」。");
    }

    private async Task<NodeRunLog> RunPluginAsync(
        LoadedPlugin plugin,
        WorkflowNode node,
        Dictionary<string, object?> inputs,
        CancellationToken cancellationToken)
    {
        Dictionary<string, object?> declared = FilterDeclaredInputs(plugin.Manifest, inputs);
        node.Config.TryGetValue("credentialId", out string? credentialId);
        Dictionary<string, object?> outputs = await plugins.RunAsync(plugin, declared, credentialId, cancellationToken);
        string? stderr = null;
        if (outputs.Remove("_stderr", out object? stderrValue))
        {
            stderr = stderrValue?.ToString();
        }

        return new NodeRunLog
        {
            NodeId = node.Id,
            Type = node.Type,
            Title = node.Title,
            Message = stderr,
            Inputs = declared,
            Outputs = outputs,
        };
    }

    private static NodeRunLog ToLog(WorkflowNode node, AgentStepResult step)
        => new()
        {
            NodeId = node.Id,
            Type = node.Type,
            Title = node.Title,
            Message = step.Message,
            Inputs = step.Inputs,
            Outputs = step.Outputs,
        };

    private static Dictionary<string, object?> FilterDeclaredInputs(PluginManifest manifest, Dictionary<string, object?> inputs)
    {
        Dictionary<string, object?> declared = new(StringComparer.OrdinalIgnoreCase);
        foreach (PluginPort port in manifest.Inputs)
        {
            if (inputs.TryGetValue(port.Name, out object? value))
            {
                declared[port.Name] = value;
            }
        }

        return declared;
    }

    private static void Validate(WorkflowGraph graph)
    {
        if (graph.Nodes.Count == 0)
        {
            throw new InvalidOperationException("画布上还没有节点。");
        }

        HashSet<string> ids = graph.Nodes.Select(n => n.Id).ToHashSet();
        foreach (WorkflowEdge edge in graph.Edges)
        {
            if (!ids.Contains(edge.From) || !ids.Contains(edge.To))
            {
                throw new InvalidOperationException("存在指向已删除节点的连线。");
            }
        }
    }

    private static List<string> TopologicalOrder(WorkflowGraph graph)
    {
        Dictionary<string, int> indegree = graph.Nodes.ToDictionary(n => n.Id, _ => 0);
        Dictionary<string, List<string>> outgoing = graph.Nodes.ToDictionary(n => n.Id, _ => new List<string>());
        foreach (WorkflowEdge edge in graph.Edges)
        {
            indegree[edge.To]++;
            outgoing[edge.From].Add(edge.To);
        }

        Queue<string> ready = new(indegree.Where(kv => kv.Value == 0).Select(kv => kv.Key));
        List<string> order = [];
        while (ready.Count > 0)
        {
            string id = ready.Dequeue();
            order.Add(id);
            foreach (string next in outgoing[id].Distinct())
            {
                indegree[next]--;
                if (indegree[next] == 0)
                {
                    ready.Enqueue(next);
                }
            }
        }

        if (order.Count != graph.Nodes.Count)
        {
            throw new InvalidOperationException("工作流存在环，无法运行。");
        }

        return order;
    }

    private static Dictionary<string, object?> ResolveInputs(
        WorkflowGraph graph,
        WorkflowNode node,
        Dictionary<string, Dictionary<string, object?>> outputs)
    {
        Dictionary<string, object?> inputs = new(StringComparer.OrdinalIgnoreCase);
        foreach (KeyValuePair<string, string> item in node.Config)
        {
            if (ReservedConfigKeys.Contains(item.Key) || string.IsNullOrWhiteSpace(item.Value))
            {
                continue;
            }

            inputs[item.Key] = item.Value;
        }

        foreach (WorkflowEdge edge in graph.Edges.Where(e => e.To == node.Id))
        {
            if (!outputs.TryGetValue(edge.From, out Dictionary<string, object?>? source))
            {
                continue;
            }

            if (source.TryGetValue(edge.FromPort, out object? value))
            {
                inputs[edge.ToPort] = value;
            }
        }

        return inputs;
    }

    private static bool PassEdgeConditions(
        WorkflowGraph graph,
        WorkflowNode node,
        Dictionary<string, Dictionary<string, object?>> outputs,
        out string? skipReason)
    {
        List<WorkflowEdge> incoming = graph.Edges.Where(e => e.To == node.Id).ToList();
        if (incoming.Count == 0)
        {
            skipReason = null;
            return true;
        }

        foreach (WorkflowEdge edge in incoming)
        {
            if (string.IsNullOrWhiteSpace(edge.When))
            {
                continue;
            }

            if (!outputs.TryGetValue(edge.From, out Dictionary<string, object?>? source))
            {
                skipReason = $"上一节点 {edge.From} 尚未产出结果。";
                return false;
            }

            bool hasCities = ReadBool(source, "hasValidCities");
            bool pass = edge.When switch
            {
                "hasValidCities" => hasCities,
                "!hasValidCities" => !hasCities,
                _ => true,
            };
            if (!pass)
            {
                skipReason = $"连线条件 {edge.When} 不满足，跳过本节点。";
                return false;
            }
        }

        skipReason = null;
        return true;
    }

    private static bool ReadBool(Dictionary<string, object?> map, string key)
    {
        if (!map.TryGetValue(key, out object? value) || value is null)
        {
            return false;
        }

        return value switch
        {
            bool b => b,
            JsonElement el when el.ValueKind == JsonValueKind.True => true,
            JsonElement el when el.ValueKind == JsonValueKind.False => false,
            _ => bool.TryParse(value.ToString(), out bool parsed) && parsed,
        };
    }

    private static string? ReadString(Dictionary<string, object?> map, string key)
    {
        if (!map.TryGetValue(key, out object? value) || value is null)
        {
            return null;
        }

        if (value is JsonElement el)
        {
            return el.ValueKind == JsonValueKind.String ? el.GetString() : el.ToString();
        }

        return value.ToString();
    }
}