XFEWorkflow

【WinUI】XFE基于windows的工作流

公开

关注 0 Fork 0 Star 0

返回提交历史

XFEstudio/XFEWorkflow

支持触发器输入映射、工作流完成联动与 XFE 运行日志持久化

b635a85

XFE工作室室长 <mail@xfegzs.com>

提交于 2026-10-11 17:47

父提交 95970c2

代码差异

8 个文件 +356 -21

Modified XFEWorkflow/XFEWorkflow.Abstractions/WorkflowModels.cs +7 -1

@@ -8,7 +8,7 @@ public enum WorkflowKind { Graph, CSharp }

8 8 public enum PortKind { Execution, Data }

9 9 public enum WorkflowValueType { Any, String, Number, Boolean, Object, Array, Resource }

10 10 public enum RunStatus { Pending, Running, Succeeded, Failed, Cancelled, Interrupted, Skipped }

11 public enum TriggerKind { Manual, Interval, Daily, FileChanged, Plugin }

11 public enum TriggerKind { Manual, Interval, Daily, FileChanged, Plugin, WorkflowCompleted }

12 12 public enum RunEventKind { Started, NodeStarted, NodeSucceeded, NodeFailed, Log, Finished }

13 13

14 14 public sealed class WorkflowDefinition

@@ -84,6 +84,9 @@ public sealed class TriggerDefinition

84 84 public string Filter { get; set; } = "*";

85 85 public string CapabilityId { get; set; } = "";

86 86 public JsonObject Parameters { get; set; } = [];

87 public JsonObject InputBindings { get; set; } = [];

88 public string SourceWorkflowId { get; set; } = "";

89 public RunStatus? CompletionStatus { get; set; } = RunStatus.Succeeded;

87 90 }

88 91

89 92 public sealed class RunRecord

@@ -97,6 +100,9 @@ public sealed class RunRecord

97 100 public DateTimeOffset? FinishedAt { get; set; }

98 101 public string? Error { get; set; }

99 102 public string? ErrorNodeId { get; set; }

103 public JsonObject Inputs { get; set; } = [];

104 public List<string> WorkflowChain { get; set; } = [];

105 public Dictionary<string, string> NodeNames { get; set; } = [];

100 106 public JsonObject Outputs { get; set; } = [];

101 107 public List<RunEvent> Events { get; set; } = [];

102 108 [JsonIgnore] public string Summary => $"{StartedAt.ToLocalTime():MM-dd HH:mm:ss} {WorkflowName} · {Status}";

Modified XFEWorkflow/XFEWorkflow.Infrastructure/Execution/RunCoordinator.cs +9 -3

@@ -50,9 +50,15 @@ public sealed class RunCoordinator(WorkspacePaths paths, PluginManager plugins,

50 50 return result.OrderByDescending(x => x.StartedAt).ToList();

51 51 }

52 52

    private Task SaveAsync(RunRecord record, CancellationToken ct = default) => AtomicJson.WriteAsync(WorkspacePaths.Inside(paths.Runs, record.Id + ".json"), record, ct);

    public async Task<RunRecord> RunAsync(WorkflowDefinition workflow, string trigger = "手动", JsonObject? inputs = null, CancellationToken ct = default)

    public async Task<RunRecord> RunAsync(WorkflowDefinition workflow, string trigger = "手动", JsonObject? inputs = null, CancellationToken ct = default, IReadOnlyList<string>? workflowChain = null)

54 54 {

        var record = new RunRecord { WorkflowId = workflow.Id, WorkflowName = workflow.Name, Trigger = trigger, Status = RunStatus.Pending };

55 var runInputs = WorkflowJson.Copy(workflow.InputDefaults);

56 foreach (var pair in inputs ?? []) runInputs[pair.Key] = pair.Value?.DeepClone();

        var record = new RunRecord { WorkflowId = workflow.Id, WorkflowName = workflow.Name, Trigger = trigger, Status = RunStatus.Pending,

58 Inputs = Sanitize(runInputs), WorkflowChain = [.. workflowChain ?? [], workflow.Id] };

59 void CaptureNames(GraphDefinition graph)

        { foreach (var node in graph.Nodes) { record.NodeNames[node.Id] = node.Name; foreach (var child in node.Children.Values) CaptureNames(child); } }

61 CaptureNames(workflow.Graph);

56 62 var session = new RunSession(record.Id);

57 63 if (!active.TryAdd(workflow.Id, session))

58 64 {

@@ -105,7 +111,7 @@ public sealed class RunCoordinator(WorkspacePaths paths, PluginManager plugins,

105 111 session.Peer.Start();

106 112 var response = await session.Peer.RequestAsync<RunResponse>("run", new RunRequest

107 113 {

108

                Workflow = workflow, RunId = record.Id, WorkspaceRoot = paths.Root, Plugins = required, Inputs = inputs ?? WorkflowJson.Copy(workflow.InputDefaults)

114

                Workflow = workflow, RunId = record.Id, WorkspaceRoot = paths.Root, Plugins = required, Inputs = runInputs

109 115 }, linked.Token) ?? throw new IOException("Runner 未返回执行结果");

110 116

            record.Status = response.Status; record.Outputs = Sanitize(response.Outputs); record.Error = response.Error is null ? null : redactor.Redact(response.Error); record.ErrorNodeId = response.NodeId;

111 117 }

Added XFEWorkflow/XFEWorkflow.Infrastructure/Execution/TriggerInputs.cs +47 -0

@@ -0,0 +1,47 @@

1 using System.Text.Json.Nodes;

2 using XFEWorkflow.Abstractions;

4 namespace XFEWorkflow.Infrastructure.Execution;

6 public static class TriggerInputs

7 {

8 public static string Label(TriggerDefinition trigger) => trigger.Kind switch

9 {

10 TriggerKind.Interval => $"每 {trigger.IntervalSeconds} 秒",

11 TriggerKind.Daily => "每日 " + trigger.DailyTime,

12 TriggerKind.FileChanged => "文件变化 · " + trigger.Filter,

13 TriggerKind.Plugin => "插件事件 · " + trigger.CapabilityId,

14 TriggerKind.WorkflowCompleted => "工作流完成",

15 _ => "手动运行"

16 };

    public static JsonObject Build(WorkflowDefinition workflow, TriggerDefinition trigger, JsonObject? data, DateTimeOffset firedAt)

19 {

20 var input = WorkflowJson.Copy(workflow.InputDefaults);

21 var payload = WorkflowJson.Copy(data);

22 foreach (var pair in payload) input[pair.Key] = pair.Value?.DeepClone();

23 foreach (var pair in trigger.InputBindings) input[pair.Key] = Resolve(pair.Value, payload);

24 input["trigger"] = new JsonObject

25 {

26 ["id"] = trigger.Id, ["kind"] = trigger.Kind.ToString(), ["name"] = Label(trigger),

27 ["firedAt"] = firedAt, ["data"] = payload

28 };

29 return input;

30 }

32 private static JsonNode? Resolve(JsonNode? value, JsonObject data)

33 {

34 if (value is JsonObject obj)

35 {

36 if (obj.Count == 1 && obj["$event"] is { } reference)

37 return WorkflowJson.ReadPath(data, reference.Text())?.DeepClone();

38 var result = new JsonObject();

39 foreach (var pair in obj) result[pair.Key] = Resolve(pair.Value, data);

40 return result;

41 }

42 if (value is JsonArray array) return new JsonArray(array.Select(x => Resolve(x, data)).ToArray());

43 return value?.DeepClone();

44 }

45 }

public sealed record TriggerScheduleStatus(string TriggerId, string Label, bool Enabled, DateTimeOffset? NextRun, string? Error);

Modified XFEWorkflow/XFEWorkflow.Infrastructure/Execution/WorkflowScheduler.cs +42 -14

@@ -1,5 +1,6 @@

1 1 using System.Collections.Concurrent;

2 2 using System.Text.Json.Nodes;

3 using System.Text.Json;

3 4 using XFEWorkflow.Abstractions;

4 5 using XFEWorkflow.Infrastructure.Storage;

5 6

@@ -13,15 +14,25 @@ public sealed class WorkflowScheduler(WorkflowRepository repository, RunCoordina

13 14 private readonly List<Task> pluginWorkers = [];

14 15 private readonly ConcurrentDictionary<string, DateTimeOffset> next = [];

15 16 private readonly ConcurrentDictionary<string, long> fileDebounce = [];

17 private readonly ConcurrentDictionary<string, string> signatures = [];

18 private readonly ConcurrentDictionary<string, string> failures = [];

19 private readonly ConcurrentDictionary<string, DateTimeOffset> completed = [];

16 20 private readonly SemaphoreSlim refreshGate = new(1);

17 21 private List<WorkflowDefinition> workflows = [];

18 22 private Task? loop;

19 23 private volatile bool suspended;

20 24 public bool IsSuspended => suspended;

21 public void Suspend() => suspended = true;

22 public async Task ResumeAsync() { await RefreshAsync(); suspended = false; }

25 public event Action? Changed;

26 public void Suspend() { suspended = true; Changed?.Invoke(); }

27 public async Task ResumeAsync() { await RefreshAsync(); suspended = false; Changed?.Invoke(); }

23 28 public event Action<string>? Error;

24 public async Task StartAsync() { await RefreshAsync(); loop = LoopAsync(); }

    public async Task StartAsync() { await RefreshAsync(); coordinator.RunChanged += WorkflowFinished; loop = LoopAsync(); }

    public IReadOnlyList<TriggerScheduleStatus> StatusFor(WorkflowDefinition workflow) => workflow.Triggers.Select(trigger =>

31 {

32 var key = workflow.Id + "/" + trigger.Id;

33 var due = next.GetValueOrDefault(key, DateTimeOffset.MaxValue);

        return new TriggerScheduleStatus(trigger.Id, TriggerInputs.Label(trigger), trigger.Enabled, due == DateTimeOffset.MaxValue ? null : due, failures.GetValueOrDefault(key));

35 }).ToArray();

25 36 public async Task RefreshAsync()

26 37 {

27 38 await refreshGate.WaitAsync(lifetime.Token);

@@ -31,11 +42,15 @@ public sealed class WorkflowScheduler(WorkflowRepository repository, RunCoordina

31 42 watchers.Clear();

32 43 await StopPluginWorkersAsync();

33 44 workflows = await repository.ListAsync(lifetime.Token);

45 failures.Clear();

            var keys = workflows.SelectMany(w => w.Triggers.Where(t => t.Enabled).Select(t => w.Id + "/" + t.Id)).ToHashSet();

            foreach (var key in next.Keys.Where(k => !keys.Contains(k))) { next.TryRemove(key, out _); signatures.TryRemove(key, out _); }

34 48 foreach (var workflow in workflows)

35 49 foreach (var trigger in workflow.Triggers.Where(x => x.Enabled))

36 50 {

37 51 var key = workflow.Id + "/" + trigger.Id;

38 next.TryAdd(key, NextDue(trigger, DateTimeOffset.Now));

52 var signature = JsonSerializer.Serialize(trigger, WorkflowJson.Options);

                    if (signatures.GetValueOrDefault(key) != signature) { next[key] = NextDue(trigger, DateTimeOffset.Now); signatures[key] = signature; }

39 54 if (trigger.Kind == TriggerKind.FileChanged)

40 55 {

41 56 try

@@ -46,13 +61,13 @@ public sealed class WorkflowScheduler(WorkflowRepository repository, RunCoordina

46 61 var now = Environment.TickCount64;

47 62 if (fileDebounce.TryGetValue(key, out var old) && now - old < 500) return;

48 63 fileDebounce[key] = now;

49 _ = FireAsync(workflow, "文件变化", new() { ["path"] = path });

64 _ = FireAsync(workflow, trigger, new() { ["path"] = path });

50 65 }

51 66

                            watcher.Changed += (_, e) => Fire(e.FullPath); watcher.Created += (_, e) => Fire(e.FullPath); watcher.Renamed += (_, e) => Fire(e.FullPath);

                            watcher.Error += (_, e) => Error?.Invoke("文件监听失败：" + e.GetException().Message);

                            watcher.Error += (_, e) => Fail(workflow, trigger, "文件监听失败：" + e.GetException().Message);

53 68 watcher.EnableRaisingEvents = true; watchers.Add(watcher);

54 69 }

55 catch (Exception ex) { Error?.Invoke(workflow.Name + "：" + ex.Message); }

70 catch (Exception ex) { Fail(workflow, trigger, ex.Message); }

56 71 }

57 72 else if (trigger.Kind == TriggerKind.Plugin)

58 73 {

@@ -62,7 +77,7 @@ public sealed class WorkflowScheduler(WorkflowRepository repository, RunCoordina

62 77 }

63 78 }

64 79 }

65 finally { refreshGate.Release(); }

80 finally { refreshGate.Release(); Changed?.Invoke(); }

66 81 }

67 82 public static DateTimeOffset NextDue(TriggerDefinition trigger, DateTimeOffset now)

68 83 {

@@ -86,23 +101,35 @@ public sealed class WorkflowScheduler(WorkflowRepository repository, RunCoordina

86 101 var key = workflow.Id + "/" + trigger.Id;

87 102 if (next.GetValueOrDefault(key, DateTimeOffset.MaxValue) > DateTimeOffset.Now) continue;

88 103 next[key] = NextDue(trigger, DateTimeOffset.Now);

                        _ = FireAsync(workflow, trigger.Kind == TriggerKind.Daily ? "每日定时" : "间隔定时", null);

104 _ = FireAsync(workflow, trigger, new() { ["scheduledAt"] = DateTimeOffset.Now });

105 Changed?.Invoke();

90 106 }

91 107 }

92 108 catch (OperationCanceledException) when (lifetime.IsCancellationRequested) { }

93 109 }

94 110

    private async Task RunPluginTriggerAsync(WorkflowDefinition workflow, TriggerDefinition trigger, CancellationToken ct)

95 111 {

        try { await coordinator.StartTriggerAsync(workflow, trigger, input => new ValueTask(FireAsync(workflow, "插件触发", input)), ct); }

112

        try { await coordinator.StartTriggerAsync(workflow, trigger, input => new ValueTask(FireAsync(workflow, trigger, input)), ct); }

97 113 catch (OperationCanceledException) when (ct.IsCancellationRequested) { }

98 catch (Exception ex) { Error?.Invoke("插件触发器已停止：" + ex.Message); }

114 catch (Exception ex) { Fail(workflow, trigger, "插件触发器已停止：" + ex.Message); }

115 }

116 private void Fail(WorkflowDefinition workflow, TriggerDefinition trigger, string message)

117

    { failures[workflow.Id + "/" + trigger.Id] = message; Error?.Invoke(workflow.Name + " · " + TriggerInputs.Label(trigger) + "：" + message); Changed?.Invoke(); }

118 private void WorkflowFinished(RunRecord record)

119 {

120

        if (record.FinishedAt is null || record.Status is RunStatus.Skipped or RunStatus.Pending or RunStatus.Running || !completed.TryAdd(record.Id, DateTimeOffset.UtcNow)) return;

121

        if (completed.Count > 2048) foreach (var old in completed.OrderBy(x => x.Value).Take(512)) completed.TryRemove(old.Key, out _);

122 foreach (var workflow in workflows.ToArray())

123

            foreach (var trigger in workflow.Triggers.Where(t => t.Enabled && t.Kind == TriggerKind.WorkflowCompleted && t.SourceWorkflowId == record.WorkflowId && (t.CompletionStatus is null || t.CompletionStatus == record.Status)))

124

                _ = FireAsync(workflow, trigger, new() { ["workflowId"] = record.WorkflowId, ["runId"] = record.Id, ["status"] = record.Status.ToString(), ["outputs"] = record.Outputs.DeepClone(), ["error"] = record.Error }, record.WorkflowChain.Count == 0 ? [record.WorkflowId] : record.WorkflowChain);

99 125 }

100 private async Task FireAsync(WorkflowDefinition workflow, string reason, JsonObject? input)

126

    private async Task FireAsync(WorkflowDefinition workflow, TriggerDefinition trigger, JsonObject? input, IReadOnlyList<string>? chain = null)

101 127 {

102 128 if (suspended) return;

103 try { await coordinator.RunAsync(WorkflowJson.Clone(workflow), reason, input, lifetime.Token); }

129

        if (chain?.Contains(workflow.Id) == true) { Fail(workflow, trigger, "检测到工作流联动环路，已阻止重复触发"); return; }

130

        try { await coordinator.RunAsync(WorkflowJson.Clone(workflow), TriggerInputs.Label(trigger), TriggerInputs.Build(workflow, trigger, input, DateTimeOffset.UtcNow), lifetime.Token, chain); }

104 131 catch (OperationCanceledException) when (lifetime.IsCancellationRequested) { }

105 catch (Exception ex) { Error?.Invoke(ex.Message); }

132 catch (Exception ex) { Fail(workflow, trigger, ex.Message); }

106 133 }

107 134 private async Task StopPluginWorkersAsync()

108 135 {

@@ -113,6 +140,7 @@ public sealed class WorkflowScheduler(WorkflowRepository repository, RunCoordina

113 140 }

114 141 public async ValueTask DisposeAsync()

115 142 {

143 coordinator.RunChanged -= WorkflowFinished;

116 144 await lifetime.CancelAsync();

117 145 foreach (var watcher in watchers) watcher.Dispose();

118 146 await StopPluginWorkersAsync();

Added XFEWorkflow/XFEWorkflow.Infrastructure/Storage/WorkflowLogService.cs +113 -0

@@ -0,0 +1,113 @@

1 using System.Text.Json;

2 using System.Text.Json.Serialization;

3 using XFEExtension.NetCore.XFEConsole;

4 using XFEWorkflow.Abstractions;

5 using XFEWorkflow.Core.Runtime;

6 using XFEWorkflow.Infrastructure.Execution;

8 namespace XFEWorkflow.Infrastructure.Storage;

10 public sealed class WorkflowLogEntry : XFELogEntry

11 {

12 public string Id { get; set; } = Guid.NewGuid().ToString("N");

13 public string Category { get; set; } = "应用";

14 public string WorkflowId { get; set; } = "";

15 public string WorkflowName { get; set; } = "";

16 public string RunId { get; set; } = "";

17 public string? NodeId { get; set; }

18 public string NodeName { get; set; } = "";

19 public string Trigger { get; set; } = "";

20 public WorkflowDiagnostic? Diagnostic { get; set; }

21 [JsonIgnore] public string TimeText => Time.ToString("HH:mm:ss.fff");

22 [JsonIgnore] public string LevelLabel => Level.ToString().ToUpperInvariant();

    [JsonIgnore] public string SourceLabel => WorkflowName.Length == 0 ? Category : WorkflowName + (NodeName.Length == 0 ? "" : " / " + NodeName);

24 [JsonIgnore] public string SearchText => $"{LogText} {SourceLabel} {RunId} {NodeId} {Trigger}";

25 }

27 /// <summary>Runner events retain their original run record; diagnostics use a separate journal.

28 /// Both are XFE log entries and share the same filtering, rendering and export pipeline.</summary>

29 public sealed class WorkflowLogService(WorkspacePaths paths, RunCoordinator runs, SecretRedactor redactor)

30 {

31 private readonly object gate = new();

32 private readonly Dictionary<string, (int Count, bool Finished)> observed = [];

33 private readonly Dictionary<string, WorkflowLogEntry[]> live = [];

34 private readonly Dictionary<string, string> diagnosticVersions = [];

35 private readonly SemaphoreSlim journalGate = new(1);

36 public event Action? Changed;

38 public static IReadOnlyList<WorkflowLogEntry> RunEntries(RunRecord run)

39 {

40 RunEvent[] events; lock (run.Events) events = run.Events.ToArray();

41 var result = events.Select((e, i) => Entry(run, e.Message, e.Timestamp.LocalDateTime,

            e.Kind == RunEventKind.NodeFailed ? LogLevel.Error : LogLevel.Info, e.NodeId, run.Id + "/event/" + i)).ToList();

43 if (run.FinishedAt is { } finished)

44 result.Add(Entry(run, run.Error ?? "工作流执行完成：" + run.Status, finished.LocalDateTime,

                run.Status == RunStatus.Failed ? LogLevel.Error : run.Status is RunStatus.Cancelled or RunStatus.Interrupted or RunStatus.Skipped ? LogLevel.Warning : LogLevel.Info,

46 run.ErrorNodeId, run.Id + "/finished"));

47 return result;

48 }

    private static WorkflowLogEntry Entry(RunRecord run, string message, DateTime time, LogLevel level, string? node, string id) => new()

50 {

        Id = id, Category = "工作流运行", Time = time, Level = level, LogText = message, WorkflowId = run.WorkflowId,

        WorkflowName = run.WorkflowName, RunId = run.Id, NodeId = node, NodeName = node is null ? "" : run.NodeNames.GetValueOrDefault(node, node), Trigger = run.Trigger

53 };

54 public void Observe(RunRecord run)

55 {

56 lock (gate)

57 {

58 var previous = observed.GetValueOrDefault(run.Id);

59 var entries = RunEntries(run);

60 live[run.Id] = entries.ToArray();

61 var eventCount = entries.Count - (run.FinishedAt is null ? 0 : 1);

            foreach (var entry in entries.Skip(previous.Count).Take(Math.Max(0, eventCount - previous.Count))) Publish(entry);

63 if (run.FinishedAt is not null && !previous.Finished) Publish(entries[^1]);

64 observed[run.Id] = (eventCount, run.FinishedAt is not null);

            if (observed.Count > 2048) foreach (var id in observed.Where(x => x.Value.Finished).Take(512).Select(x => x.Key).ToArray()) { observed.Remove(id); live.Remove(id); }

66 }

67 Changed?.Invoke();

68 }

69 private void Publish(WorkflowLogEntry entry)

70 { entry.LogText = redactor.Redact(entry.LogText); XFEConsole.Log.AddLog(entry); }

    public async Task<IReadOnlyList<WorkflowLogEntry>> DiagnosticsAsync(WorkflowDefinition workflow, IReadOnlyList<WorkflowDiagnostic> diagnostics, CancellationToken ct = default)

73 {

74 var entries = diagnostics.Select(d => new WorkflowLogEntry

75 {

76 Category = "诊断", WorkflowId = workflow.Id, WorkflowName = workflow.Name, NodeId = d.NodeId,

77 NodeName = d.NodeId ?? d.File ?? "编译与校验", Level = d.IsError ? LogLevel.Error : LogLevel.Warning,

78 LogText = redactor.Redact(d.Message), Diagnostic = d with { Message = redactor.Redact(d.Message) }

79 }).ToArray();

80 await journalGate.WaitAsync(ct);

81 try

82 {

83 var signature = JsonSerializer.Serialize(diagnostics, WorkflowJson.Options);

84 if (diagnosticVersions.GetValueOrDefault(workflow.Id) == signature) return entries;

85 diagnosticVersions[workflow.Id] = signature;

86 if (entries.Length == 0) return entries;

87 var path = WorkspacePaths.Inside(paths.Logs, workflow.Id + ".diagnostics.json");

88 var previous = await AtomicJson.ReadAsync<List<WorkflowLogEntry>>(path, ct) ?? [];

89 await AtomicJson.WriteAsync(path, previous.Concat(entries).TakeLast(5000).ToList(), ct);

90 lock (gate) foreach (var entry in entries) Publish(entry);

91 }

92 finally { journalGate.Release(); }

93 Changed?.Invoke(); return entries;

94 }

95 public async Task<List<WorkflowLogEntry>> HistoryAsync(CancellationToken ct = default)

96 {

97 var entries = (await runs.HistoryAsync(ct)).SelectMany(RunEntries).ToList();

98 lock (gate) entries.AddRange(live.Values.SelectMany(x => x));

99 foreach (var file in Directory.EnumerateFiles(paths.Logs, "*.diagnostics.json"))

100 {

101 try { entries.AddRange(await AtomicJson.ReadAsync<List<WorkflowLogEntry>>(file, ct) ?? []); }

102 catch (JsonException) { }

103 }

104 // Workflow entries in the XFE console are already represented by their

105 // durable run record or diagnostic journal. Do not duplicate them.

106 XFELogEntry[] application;

107 lock (gate) application = XFEConsole.Log.Logs.ToArray();

108 entries.AddRange(application.Select((x, i) => (Entry: x, Index: i)).Where(x => x.Entry is not WorkflowLogEntry)

109

            .Select(x => new WorkflowLogEntry { Id = "application/" + x.Entry.Time.Ticks + "/" + x.Index, Time = x.Entry.Time, Level = x.Entry.Level, LogText = redactor.Redact(x.Entry.LogText) }));

110 foreach (var entry in entries) entry.LogText = redactor.Redact(entry.LogText);

111 return entries.DistinctBy(x => x.Id).OrderByDescending(x => x.Time).ToList();

112 }

113 }

Modified XFEWorkflow/XFEWorkflow.Infrastructure/Storage/WorkspaceStorage.cs +7 -3

@@ -13,10 +13,11 @@ public sealed class WorkspacePaths

13 13 public string Runs => Path.Combine(Root, "Runs");

14 14 public string Temp => Path.Combine(Root, "Temp");

15 15 public string BrowserProfiles => Path.Combine(Root, "BrowserProfiles");

16 public string Logs => Path.Combine(Root, "Logs");

16 17 public WorkspacePaths(string root)

17 18 {

18 19 Root = Path.GetFullPath(root);

        foreach (var directory in new[] { Root, Workflows, Plugins, Runs, Temp, BrowserProfiles }) Directory.CreateDirectory(directory);

        foreach (var directory in new[] { Root, Workflows, Plugins, Runs, Temp, BrowserProfiles, Logs }) Directory.CreateDirectory(directory);

20 21 }

21 22 public static string Inside(string root, string relative)

22 23 {

@@ -131,7 +132,6 @@ public sealed class WorkflowRepository(WorkspacePaths paths)

131 132 public partial class ClientStateProfile : XFEProfile

132 133 {

133 134 [ProfileProperty] private List<InstalledPlugin> _plugins = [];

134 [ProfileProperty] private string _spaceAddress = "http://localhost:3300/space/v1/";

135 135 [ProfileProperty] private string _deviceId = Guid.NewGuid().ToString("N");

136 136 [ProfileProperty] private string _desktopBackend = "standard";

137 137 [ProfileProperty] private Dictionary<string, string> _credentials = [];

@@ -146,6 +146,8 @@ public partial class ClientStateProfile : XFEProfile

146 146

147 147 public sealed class ClientState : IAsyncDisposable

148 148 {

149 public const string LocalServerAddress = "http://localhost:3300/space/v1/";

150 private string serverAddress = LocalServerAddress;

149 151 public ProfileStore<ClientStateProfile> Store { get; }

150 152 public ClientState(WorkspacePaths paths)

151 153 {

@@ -153,7 +155,9 @@ public sealed class ClientState : IAsyncDisposable

153 155 Store.Load();

154 156 }

155 157 public List<InstalledPlugin> Plugins => Store.Read(p => WorkflowJson.Clone(p.InstancePlugins));

156

    public string SpaceAddress { get => Store.Read(p => p.InstanceSpaceAddress); set => Store.Update(p => p.InstanceSpaceAddress = value); }

158 // The desktop product owns its server endpoint. Persisted legacy addresses

159 // are ignored; only the internal UI fixture can substitute a loopback server.

160

    public string SpaceAddress { get => serverAddress; internal set => serverAddress = Space.SpaceClient.ValidateAddress(value); }

157 161 public string DeviceId => Store.Read(p => p.InstanceDeviceId);

158 162

    public string DesktopBackend { get => Store.Read(p => p.InstanceDesktopBackend); set => Store.Update(p => p.InstanceDesktopBackend = value); }

159 163

    public double PaneSize(string key, double fallback) => Store.Read(p => p.InstancePaneSizes.TryGetValue(key, out var value) && double.IsFinite(value) && value > 0 ? value : fallback);

Modified XFEWorkflow/XFEWorkflow.Infrastructure/XFEWorkflow.Infrastructure.csproj +2 -0

@@ -7,6 +7,7 @@

7 7 <PackageReference Include="Microsoft.CodeAnalysis.CSharp.Features" Version="5.6.0" />

8 8 <PackageReference Include="System.Security.Cryptography.ProtectedData" Version="10.0.12" />

9 9 <PackageReference Include="XFEExtension.NetCore.ServerInteractive" Version="4.0.1" />

10 <PackageReference Include="XFEExtension.NetCore.XFEConsole" Version="2.6.0" />

10 11 <PackageDownload Include="Microsoft.NETCore.App.Ref" Version="[10.0.12]" />

11 12 </ItemGroup>

12 13 <ItemGroup>

@@ -14,4 +15,5 @@

14 15 Link="CompilerReferences\%(Filename)%(Extension)"

15 16 CopyToOutputDirectory="PreserveNewest" CopyToPublishDirectory="PreserveNewest" />

16 17 </ItemGroup>

18 <ItemGroup><InternalsVisibleTo Include="XFEWorkflow" /><InternalsVisibleTo Include="XFEWorkflow.Test" /></ItemGroup>

17 19 </Project>

Added XFEWorkflow/XFEWorkflow.Test/WorkbenchTests.cs +129 -0

@@ -0,0 +1,129 @@

1 using System.Text.Json.Nodes;

2 using XFEExtension.NetCore.XFEConsole;

3 using XFEExtension.NetCore.XUnit.Assertions;

4 using XFEExtension.NetCore.XUnit.Attributes;

5 using XFEWorkflow.Abstractions;

6 using XFEWorkflow.Core.Runtime;

7 using XFEWorkflow.Infrastructure.Execution;

8 using XFEWorkflow.Infrastructure.Plugins;

9 using XFEWorkflow.Infrastructure.Storage;

11 namespace XFEWorkflow.Test;

13 [TestFixture]

14 public class WorkbenchTests

15 {

    private static RunCoordinator Coordinator(TestWorkspace workspace, SecretRedactor? redactor = null) => new(workspace.Paths,

        new PluginManager(workspace.Paths, workspace.State), new CredentialVault(workspace.State), redactor ?? new(), TestSupport.FindOutput("XFEWorkflow.Runner", "XFEWorkflow.Runner.exe"));

19 [Test]

20 public void TriggerBindingsPreserveDefaultsAndCopyNestedEventValues()

21 {

22 var workflow = TestSupport.Graph(); workflow.InputDefaults = new() { ["default"] = "keep", ["value"] = "old" };

23 var trigger = new TriggerDefinition { Kind = TriggerKind.WorkflowCompleted, InputBindings = new()

24 {

25 ["value"] = new JsonObject { ["$event"] = "outputs.result" },

26 ["nested"] = new JsonArray(new JsonObject { ["$event"] = "outputs.result" }, "constant"),

27 ["missing"] = new JsonObject { ["$event"] = "unknown" }

28 } };

        var payload = new JsonObject { ["outputs"] = new JsonObject { ["result"] = "联动输出" }, ["eventValue"] = 42 };

30 var input = TriggerInputs.Build(workflow, trigger, payload, DateTimeOffset.UtcNow);

        Assert.Equal("keep", input.String("default")); Assert.Equal("联动输出", input.String("value")); Assert.Equal(42, input.Integer("eventValue"));

32 Assert.Equal("联动输出", input["nested"]![0]!.GetValue<string>()); Assert.True(input["missing"] is null);

33 Assert.Equal(trigger.Id, input["trigger"]!["id"]!.GetValue<string>());

34 payload["outputs"]!["result"] = "changed";

35 Assert.Equal("联动输出", input["trigger"]!["data"]!["outputs"]!["result"]!.GetValue<string>());

36 Assert.Equal("old", workflow.InputDefaults.String("value"));

37 }

39 [Test]

40 public async Task RunnerMergesExplicitInputsWithWorkflowDefaultsAndPersistsInputs()

41 {

42 await using var workspace = new TestWorkspace(); await using var coordinator = Coordinator(workspace);

43 var workflow = TestSupport.Graph(); workflow.InputDefaults = new() { ["default"] = "keep", ["value"] = "old" };

44 workflow.Graph.Nodes[^1].Parameters["result"] = new JsonObject { ["$input"] = "default" };

45 var run = await coordinator.RunAsync(workflow, inputs: new() { ["value"] = "new" });

46 Assert.Equal(RunStatus.Succeeded, run.Status, run.Error); Assert.Equal("keep", run.Outputs.String("result"));

47 Assert.Equal("new", run.Inputs.String("value")); Assert.Equal(workflow.Id, run.WorkflowChain.Single());

        var saved = (await coordinator.HistoryAsync()).Single(); Assert.Equal("keep", saved.Inputs.String("default")); Assert.Equal(workflow.Graph.Nodes[0].Name, saved.NodeNames[workflow.Graph.Nodes[0].Id]);

49 }

51 [Test]

52 public async Task CompletionTriggerMapsOutputsAndBlocksCycles()

53 {

54 await using var workspace = new TestWorkspace(); await using var coordinator = Coordinator(workspace);

        var source = TestSupport.Graph(); source.Name = "上游"; source.Graph.Nodes[^1].Parameters["result"] = "upstream";

        var target = TestSupport.Graph(); target.Name = "下游"; target.Graph.Nodes[^1].Parameters["result"] = new JsonObject { ["$input"] = "message" };

57 target.Triggers.Add(new() { Kind = TriggerKind.WorkflowCompleted, Enabled = true, SourceWorkflowId = source.Id,

58 InputBindings = new() { ["message"] = new JsonObject { ["$event"] = "outputs.result" } } });

        source.Triggers.Add(new() { Kind = TriggerKind.WorkflowCompleted, Enabled = true, SourceWorkflowId = target.Id });

        var repository = new WorkflowRepository(workspace.Paths); await repository.SaveAsync(source); await repository.SaveAsync(target);

61 await using var scheduler = new WorkflowScheduler(repository, coordinator);

62 var finished = new TaskCompletionSource<RunRecord>(TaskCreationOptions.RunContinuationsAsynchronously);

63 var cycle = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);

64 scheduler.Error += message => { if (message.Contains("环路")) cycle.TrySetResult(message); };

        coordinator.RunChanged += run => { if (run.WorkflowId == target.Id && run.Status == RunStatus.Succeeded) finished.TrySetResult(run); };

        await scheduler.StartAsync(); var first = await coordinator.RunAsync(source); Assert.Equal(RunStatus.Succeeded, first.Status, first.Error);

        var linked = await finished.Task.WaitAsync(TimeSpan.FromSeconds(15)); await cycle.Task.WaitAsync(TimeSpan.FromSeconds(5));

        Assert.Equal("upstream", linked.Outputs.String("result")); Assert.Equal(first.Id, linked.Inputs["trigger"]!["data"]!["runId"]!.GetValue<string>());

69 Assert.True(linked.WorkflowChain.SequenceEqual(new[] { source.Id, target.Id }));

70 Assert.Equal(2, (await coordinator.HistoryAsync()).Count);

71 Assert.True(scheduler.StatusFor(source).Single().Error!.Contains("环路"));

72 }

74 [Test]

75 public async Task CompletionStatusFilterDoesNotRunOnFailure()

76 {

77 await using var workspace = new TestWorkspace(); await using var coordinator = Coordinator(workspace);

        var source = TestSupport.Graph(new NodeDefinition { CapabilityId = "core.csharp", Code = "throw new InvalidOperationException(\"expected\");" });

        var target = TestSupport.Graph(); target.Triggers.Add(new() { Kind = TriggerKind.WorkflowCompleted, Enabled = true, SourceWorkflowId = source.Id, CompletionStatus = RunStatus.Succeeded });

        var repository = new WorkflowRepository(workspace.Paths); await repository.SaveAsync(source); await repository.SaveAsync(target);

81 await using var scheduler = new WorkflowScheduler(repository, coordinator); await scheduler.StartAsync();

        var record = await coordinator.RunAsync(source); Assert.Equal(RunStatus.Failed, record.Status); Assert.Equal(1, (await coordinator.HistoryAsync()).Count);

83 }

85 [Test]

86 public async Task EditingIntervalResetsNextExecutionButUnchangedRefreshPreservesIt()

87 {

88 await using var workspace = new TestWorkspace(); await using var coordinator = Coordinator(workspace);

        var workflow = TestSupport.Graph(); workflow.Triggers.Add(new() { Kind = TriggerKind.Interval, Enabled = true, IntervalSeconds = 600 });

90 var repository = new WorkflowRepository(workspace.Paths); await repository.SaveAsync(workflow);

91 await using var scheduler = new WorkflowScheduler(repository, coordinator); await scheduler.StartAsync();

        var original = scheduler.StatusFor(workflow).Single().NextRun; await scheduler.RefreshAsync(); Assert.Equal(original, scheduler.StatusFor(workflow).Single().NextRun);

93 workflow.Triggers[0].IntervalSeconds = 30; await repository.SaveAsync(workflow); await scheduler.RefreshAsync();

94 Assert.True(scheduler.StatusFor(workflow).Single().NextRun < original!.Value.AddMinutes(-8));

        workflow.Triggers[0].Enabled = false; await repository.SaveAsync(workflow); await scheduler.RefreshAsync(); Assert.True(scheduler.StatusFor(workflow).Single().NextRun is null);

96 }

98 [Test]

99 public async Task WorkflowLogsAreXfeEntriesVisibleLiveDurableAndRedacted()

100 {

101

        await using var workspace = new TestWorkspace(); var redactor = new SecretRedactor(); redactor.Add("test-secret-value");

102

        await using var coordinator = Coordinator(workspace, redactor); var logs = new WorkflowLogService(workspace.Paths, coordinator, redactor); coordinator.RunChanged += logs.Observe;

103

        var step = new NodeDefinition { CapabilityId = "core.log", Name = "日志节点", Parameters = new() { ["message"] = "日志 test-secret-value" } };

104

        var workflow = TestSupport.Graph(step, new() { CapabilityId = "core.delay", Parameters = new() { ["milliseconds"] = 800 } });

105 var started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);

106

        coordinator.RunChanged += run => { if (run.Events.Any(e => e.Message.Contains("日志 "))) started.TrySetResult(); };

107 var running = coordinator.RunAsync(workflow); await started.Task.WaitAsync(TimeSpan.FromSeconds(15));

108 var live = (await logs.HistoryAsync()).Where(e => e.WorkflowId == workflow.Id).ToArray();

109

        Assert.True(live.Any(e => e.NodeName == "日志节点" && e.LogText.Contains("日志 "))); Assert.True(live.All(e => e is XFELogEntry && !e.LogText.Contains("test-secret-value")));

110 var result = await running; Assert.Equal(RunStatus.Succeeded, result.Status, result.Error);

111 await logs.DiagnosticsAsync(workflow, [new("诊断 test-secret-value", step.Id)]);

112

        var reopened = new WorkflowLogService(workspace.Paths, coordinator, redactor); var history = (await reopened.HistoryAsync()).Where(e => e.WorkflowId == workflow.Id).ToList();

113

        Assert.True(history.Any(e => e.RunId == result.Id && e.NodeId == step.Id)); Assert.True(history.Any(e => e.Category == "诊断" && e.Diagnostic is not null));

114

        Assert.True(history.All(e => !e.LogText.Contains("test-secret-value"))); Assert.Equal(history.Count, history.Select(e => e.Id).Distinct().Count());

115 coordinator.RunChanged -= logs.Observe;

116 }

117

118 [Test]

119 public async Task LoginAndSpaceEndpointIsFixedAndCannotBeSetPublicly()

120 {

121 await using var workspace = new TestWorkspace();

122 Assert.Equal("http://localhost:3300/space/v1/", workspace.State.SpaceAddress);

123 Assert.False(typeof(ClientState).GetProperty(nameof(ClientState.SpaceAddress))!.SetMethod!.IsPublic);

124

        Assert.False(typeof(ClientState).GetProperties().Any(x => x.Name.Contains("InstanceSpaceAddress", StringComparison.Ordinal)));

125 await workspace.State.SaveAsync();

126 var text = await File.ReadAllTextAsync(Path.Combine(workspace.Paths.Root, "ClientState.xpf"));

127 Assert.False(text.Contains("spaceAddress", StringComparison.OrdinalIgnoreCase));

128 }

129 }