From e705d3c612e4964dde4fb263a04c91d821d50021 Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:04:47 +0800 Subject: [PATCH 01/10] Synchronize shared runtime variables across parallel nodes --- main.py | 54 ++++++++++++++++++++++++++++++++---------------------- 1 file changed, 32 insertions(+), 22 deletions(-) diff --git a/main.py b/main.py index 4f5024f..a951c4c 100644 --- a/main.py +++ b/main.py @@ -558,7 +558,7 @@ def clear(self): class AutomationCore: def __init__(self, log_callback, app_instance): self.running = False; self.paused = False; self.stop_event = threading.Event(); self.pause_event = threading.Event() - self.log = log_callback; self.app = app_instance; self.project = None; self.runtime_memory = {}; self.io_lock = threading.Lock() + self.log = log_callback; self.app = app_instance; self.project = None; self.runtime_memory = {}; self.memory_lock = threading.RLock(); self.io_lock = threading.Lock() self.active_threads = 0; self.thread_lock = threading.Lock(); self.scaling_ratio = 1.0; self.breakpoints = set() self.max_threads = 50 self.pending_nodes = [] @@ -602,12 +602,13 @@ def start(self, start_node_id=None): self.pause_event.set() self.active_threads = 0 self.pending_nodes.clear() - self.runtime_memory = {} - self.context = {'window_rect': None, 'window_handle': 0, 'window_offset': (0, 0)} - self.performance_stats = {'nodes_executed': 0, 'errors': 0, 'start_time': time.time()} - self.log("🚀 引擎启动", "exec") - self.app.iconify() try: + with self.memory_lock: + self.runtime_memory.clear() + self.context = {'window_rect': None, 'window_handle': 0, 'window_offset': (0, 0)} + self.performance_stats = {'nodes_executed': 0, 'errors': 0, 'start_time': time.time()} + self.log("🚀 引擎启动", "exec") + self.app.iconify() threading.Thread(target=self._run_flow_engine, args=(start_node_id,), daemon=True).start() except Exception as e: with self.thread_lock: @@ -755,15 +756,17 @@ def _process_node_thread(self, node_id): self.pause_event.set() self.log(f"无法启动排队节点[{next_id}]，已停止剩余任务: {e}", "error") + def _memory_snapshot(self): + with self.memory_lock: + return dict(self.runtime_memory) + def _replace_variables(self, text): if not isinstance(text, str): return str(text) - try: - for k, v in self.runtime_memory.items(): - val_str = str(v) if v is not None else "" - text = text.replace(f'${{{k}}}', val_str) - text = text.replace(f'${k}', val_str) - text = text.replace(f'{{{k}}}', val_str) - except Exception: pass + for k, v in self._memory_snapshot().items(): + val_str = str(v) if v is not None else "" + text = text.replace(f'${{{k}}}', val_str) + text = text.replace(f'${k}', val_str) + text = text.replace(f'{{{k}}}', val_str) return text def _update_context_rect(self): @@ -816,7 +819,8 @@ def _execute_node(self, node): msg = data.get('msg', '执行到此节点') if "{ALL_VARS}" in msg.upper() or "{所有变量}" in msg or "${ALL_VARS}" in msg.upper() or "${所有变量}" in msg: - vars_list_str = "\n".join([f"• {k} = {v}" for k, v in self.runtime_memory.items()]) if self.runtime_memory else "（目前无任何内存变量）" + memory = self._memory_snapshot() + vars_list_str = "\n".join([f"• {k} = {v}" for k, v in memory.items()]) if memory else "（目前无任何内存变量）" msg = msg.replace("${ALL_VARS}", vars_list_str).replace("${all_vars}", vars_list_str).replace("${所有变量}", vars_list_str).replace("{ALL_VARS}", vars_list_str).replace("{all_vars}", vars_list_str).replace("{所有变量}", vars_list_str) use_sound = bool(data.get('use_sound', False)) @@ -867,15 +871,19 @@ def _execute_node(self, node): return 'fail' if ntype == 'set_var': - if 'batch_vars' in data: [self.runtime_memory.update({i['name']:i['value']}) for i in data['batch_vars'] if i.get('name')] - if data.get('var_name'): self.runtime_memory[data['var_name']] = data.get('var_value', '') + with self.memory_lock: + if 'batch_vars' in data: + self.runtime_memory.update({i['name']: i['value'] for i in data['batch_vars'] if i.get('name')}) + if data.get('var_name'): + self.runtime_memory[data['var_name']] = data.get('var_value', '') return 'out' if ntype == 'clipboard': mode = data.get('clip_mode', 'read') if mode == 'read': if HAS_PYPERCLIP: - self.runtime_memory[data.get('var_name', 'clipboard_data')] = pyperclip.paste() + with self.memory_lock: + self.runtime_memory[data.get('var_name', 'clipboard_data')] = pyperclip.paste() else: self.log("⚠️ 未安装 pyperclip，无法读取剪贴板", "warning") elif mode == 'write': @@ -886,10 +894,12 @@ def _execute_node(self, node): return 'out' if ntype == 'var_switch': - val = str(self.runtime_memory.get(data.get('var_name',''), '')) + with self.memory_lock: + val = str(self.runtime_memory.get(data.get('var_name',''), '')) op, target = data.get('operator', '='), data.get('var_value', '') if data.get('var_name'): return 'yes' if ((val==target) if op=='=' else (val!=target)) else 'no' - vals = [str(self.runtime_memory.get(vn.strip(), '')) for vn in data.get('var_list', '').split(',') if vn.strip()] + with self.memory_lock: + vals = [str(self.runtime_memory.get(vn.strip(), '')) for vn in data.get('var_list', '').split(',') if vn.strip()] for case in data.get('cases', []): if all(v == case.get('value', '') for v in vals): return case.get('id', 'else') return 'else' @@ -1102,11 +1112,11 @@ def _execute_node(self, node): if ntype == 'web': webbrowser.open(data.get('url')); self._smart_wait(2); return 'out' if ntype == 'loop': if data.get('infinite', True): return 'loop' - with self.io_lock: + with self.memory_lock: k = f"loop_{node['id']}"; c = self.runtime_memory.get(k, 0) if c < safe_int(data.get('count', 3)): self.runtime_memory[k] = c + 1; return 'loop' - else: - if k in self.runtime_memory: del self.runtime_memory[k] + else: + self.runtime_memory.pop(k, None) return 'exit' if ntype == 'if_img': self._ensure_window_focus() From 2552c36ba2654dfe73eca7a150399b8b966eab62 Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:05:23 +0800 Subject: [PATCH 02/10] Test startup cleanup and synchronized variable snapshots --- tests/test_regressions.py | 56 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/tests/test_regressions.py b/tests/test_regressions.py index e750885..ec98aef 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -71,6 +71,10 @@ def setUpClass(cls): "AutomationCore", "start", {"threading": threading, "time": time} ) + cls.MemorySnapshotProbe = extract_class_with_method( + "AutomationCore", "_memory_snapshot", + {"threading": threading} + ) cls.ForkProbe = extract_class_with_method( "AutomationCore", "_fork_node", {"threading": threading} @@ -130,6 +134,7 @@ def reset_ui_state(self): engine.active_threads = 0 engine.pending_nodes = [] engine.runtime_memory = {} + engine.memory_lock = threading.RLock() engine.context = {} engine.performance_stats = {} engine.app = App() @@ -147,6 +152,57 @@ def reset_ui_state(self): self.assertTrue(engine.running) self.assertEqual(FakeThread.starts, 1) + def test_start_initialization_failure_resets_running_state(self): + import threading + + class Event: + def __init__(self, set_value=False): + self.value = set_value + def clear(self): + self.value = False + def set(self): + self.value = True + + class App: + def iconify(self): + raise RuntimeError("window unavailable") + def after(self, delay, callback): + callback() + + engine = self.StartProbe() + engine.log = lambda *args: None + engine.thread_lock = threading.Lock() + engine.memory_lock = threading.RLock() + engine.running = False + engine.project = {"nodes": {"start": {"type": "start"}}} + engine.paused = False + engine.stop_event = Event() + engine.pause_event = Event(True) + engine.active_threads = 0 + engine.pending_nodes = [] + engine.runtime_memory = {"old": "value"} + engine.app = App() + engine.app.reset_ui_state = lambda: None + + with self.assertRaisesRegex(RuntimeError, "window unavailable"): + engine.start() + + self.assertFalse(engine.running) + self.assertTrue(engine.stop_event.is_set()) + self.assertTrue(engine.pause_event.is_set()) + self.assertEqual(engine.pending_nodes, []) + + def test_memory_snapshot_is_consistent_and_detached(self): + import threading + + core = self.MemorySnapshotProbe() + core.memory_lock = threading.RLock() + core.runtime_memory = {"a": 1} + snapshot = core._memory_snapshot() + core.runtime_memory["b"] = 2 + self.assertEqual(snapshot, {"a": 1}) + self.assertEqual(core._memory_snapshot(), {"a": 1, "b": 2}) + def test_fork_queues_node_when_worker_limit_is_reached(self): import threading From c745db4312bfc3bb60c8aab2f3d737442003793c Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:06:31 +0800 Subject: [PATCH 03/10] Fix event stub in startup failure regression test --- tests/test_regressions.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/test_regressions.py b/tests/test_regressions.py index ec98aef..2cb030b 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -162,6 +162,8 @@ def clear(self): self.value = False def set(self): self.value = True + def is_set(self): + return self.value class App: def iconify(self): From d47bace94d49a84fa7b0427846b6d4cf2bdf1386 Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:07:33 +0800 Subject: [PATCH 04/10] Ignore stale UI cleanup from previously completed runs --- main.py | 31 +++++++++++++++++++++++-------- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/main.py b/main.py index a951c4c..b94e919 100644 --- a/main.py +++ b/main.py @@ -558,7 +558,7 @@ def clear(self): class AutomationCore: def __init__(self, log_callback, app_instance): self.running = False; self.paused = False; self.stop_event = threading.Event(); self.pause_event = threading.Event() - self.log = log_callback; self.app = app_instance; self.project = None; self.runtime_memory = {}; self.memory_lock = threading.RLock(); self.io_lock = threading.Lock() + self.log = log_callback; self.app = app_instance; self.project = None; self.runtime_memory = {}; self.memory_lock = threading.RLock(); self.io_lock = threading.Lock(); self.run_generation = 0 self.active_threads = 0; self.thread_lock = threading.Lock(); self.scaling_ratio = 1.0; self.breakpoints = set() self.max_threads = 50 self.pending_nodes = [] @@ -597,6 +597,8 @@ def start(self, start_node_id=None): if self.running or not self.project: return self.running = True + self.run_generation += 1 + generation = self.run_generation self.paused = False self.stop_event.clear() self.pause_event.set() @@ -609,7 +611,7 @@ def start(self, start_node_id=None): self.performance_stats = {'nodes_executed': 0, 'errors': 0, 'start_time': time.time()} self.log("🚀 引擎启动", "exec") self.app.iconify() - threading.Thread(target=self._run_flow_engine, args=(start_node_id,), daemon=True).start() + threading.Thread(target=self._run_flow_engine, args=(start_node_id, generation), daemon=True).start() except Exception as e: with self.thread_lock: self.running = False @@ -657,7 +659,7 @@ def _check_pause(self, node_id=None): def _get_next_links(self, node_id, port_name='out'): return [l['target'] for l in self.project['links'] if l.get('source') == node_id and l.get('source_port', 'out') == port_name and l.get('target') in self.project['nodes']] - def _run_flow_engine(self, start_node_id=None): + def _run_flow_engine(self, start_node_id=None, generation=None): try: start_nodes = [start_node_id] if start_node_id else [nid for nid, n in self.project['nodes'].items() if n['type'] == 'start'] if not start_nodes: self.log("未找到开始节点", "error"); return @@ -688,14 +690,27 @@ def _run_flow_engine(self, start_node_id=None): if active_threads <= 0: break time.sleep(0.05) - self.running = False + with self.thread_lock: + if generation is None: + generation = self.run_generation + if generation == self.run_generation: + self.running = False if self.performance_stats['start_time']: elapsed = time.time() - self.performance_stats['start_time'] self.log(f"📊 执行统计: {self.performance_stats['nodes_executed']}个节点, {self.performance_stats['errors']}个错误, 耗时{elapsed:.2f}秒", "info") - self.log("🏁 流程结束", "info"); - self.app.highlight_node_safe(None); - self.app.after(0, self.app.deiconify); - self.app.after(100, self.app.reset_ui_state) + self.log("🏁 流程结束", "info") + self.app.after(0, lambda: self._finish_run_ui(generation, reset=False)) + self.app.after(100, lambda: self._finish_run_ui(generation, reset=True)) + + def _finish_run_ui(self, generation, reset=False): + with self.thread_lock: + if generation != self.run_generation or self.running: + return + if reset: + self.app.reset_ui_state() + else: + self.app.highlight_node_safe(None) + self.app.deiconify() def _fork_node(self, node_id): with self.thread_lock: From 1687b53f829d379bf1ed1517ac223753524b224b Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:08:10 +0800 Subject: [PATCH 05/10] Snapshot completion statistics before allowing a new run --- main.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/main.py b/main.py index b94e919..b3b04c4 100644 --- a/main.py +++ b/main.py @@ -693,11 +693,13 @@ def _run_flow_engine(self, start_node_id=None, generation=None): with self.thread_lock: if generation is None: generation = self.run_generation - if generation == self.run_generation: - self.running = False - if self.performance_stats['start_time']: - elapsed = time.time() - self.performance_stats['start_time'] - self.log(f"📊 执行统计: {self.performance_stats['nodes_executed']}个节点, {self.performance_stats['errors']}个错误, 耗时{elapsed:.2f}秒", "info") + if generation != self.run_generation: + return + stats = dict(self.performance_stats) + self.running = False + if stats['start_time']: + elapsed = time.time() - stats['start_time'] + self.log(f"📊 执行统计: {stats['nodes_executed']}个节点, {stats['errors']}个错误, 耗时{elapsed:.2f}秒", "info") self.log("🏁 流程结束", "info") self.app.after(0, lambda: self._finish_run_ui(generation, reset=False)) self.app.after(100, lambda: self._finish_run_ui(generation, reset=True)) From b9241b0de00e759ecccc2ebc8537c303285ded3b Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:08:31 +0800 Subject: [PATCH 06/10] Skip stale completion cleanup without suppressing exceptions --- main.py | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/main.py b/main.py index b3b04c4..f74a900 100644 --- a/main.py +++ b/main.py @@ -693,16 +693,17 @@ def _run_flow_engine(self, start_node_id=None, generation=None): with self.thread_lock: if generation is None: generation = self.run_generation - if generation != self.run_generation: - return - stats = dict(self.performance_stats) - self.running = False - if stats['start_time']: - elapsed = time.time() - stats['start_time'] - self.log(f"📊 执行统计: {stats['nodes_executed']}个节点, {stats['errors']}个错误, 耗时{elapsed:.2f}秒", "info") - self.log("🏁 流程结束", "info") - self.app.after(0, lambda: self._finish_run_ui(generation, reset=False)) - self.app.after(100, lambda: self._finish_run_ui(generation, reset=True)) + current_run = generation == self.run_generation + stats = dict(self.performance_stats) if current_run else None + if current_run: + self.running = False + if stats is not None: + if stats['start_time']: + elapsed = time.time() - stats['start_time'] + self.log(f"📊 执行统计: {stats['nodes_executed']}个节点, {stats['errors']}个错误, 耗时{elapsed:.2f}秒", "info") + self.log("🏁 流程结束", "info") + self.app.after(0, lambda: self._finish_run_ui(generation, reset=False)) + self.app.after(100, lambda: self._finish_run_ui(generation, reset=True)) def _finish_run_ui(self, generation, reset=False): with self.thread_lock: From aa040b19a80a5a9213969ee8e0511d7292103c0d Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:08:47 +0800 Subject: [PATCH 07/10] Test stale flow completion callbacks cannot reset a newer run --- tests/test_regressions.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/tests/test_regressions.py b/tests/test_regressions.py index 2cb030b..6e68530 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -75,6 +75,10 @@ def setUpClass(cls): "AutomationCore", "_memory_snapshot", {"threading": threading} ) + cls.FinishUiProbe = extract_class_with_method( + "AutomationCore", "_finish_run_ui", + {"threading": threading} + ) cls.ForkProbe = extract_class_with_method( "AutomationCore", "_fork_node", {"threading": threading} @@ -127,6 +131,7 @@ def reset_ui_state(self): engine.log = lambda *args: None engine.thread_lock = threading.Lock() engine.running = False + engine.run_generation = 0 engine.project = {"nodes": {"start": {"type": "start"}}} engine.paused = False engine.stop_event = Event() @@ -194,6 +199,31 @@ def after(self, delay, callback): self.assertTrue(engine.pause_event.is_set()) self.assertEqual(engine.pending_nodes, []) + def test_stale_run_cannot_reset_new_run_ui(self): + import threading + + class App: + def __init__(self): + self.calls = [] + def reset_ui_state(self): + self.calls.append("reset") + def highlight_node_safe(self, node_id): + self.calls.append(("highlight", node_id)) + def deiconify(self): + self.calls.append("show") + + core = self.FinishUiProbe() + core.thread_lock = threading.Lock() + core.run_generation = 2 + core.running = False + core.app = App() + + core._finish_run_ui(1, reset=True) + self.assertEqual(core.app.calls, []) + + core._finish_run_ui(2, reset=True) + self.assertEqual(core.app.calls, ["reset"]) + def test_memory_snapshot_is_consistent_and_detached(self): import threading From 09844ae75b05dabc289e1572c94bd8585f532c9a Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:09:35 +0800 Subject: [PATCH 08/10] Initialize run generation in startup failure test --- tests/test_regressions.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_regressions.py b/tests/test_regressions.py index 6e68530..a7532f7 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -181,6 +181,7 @@ def after(self, delay, callback): engine.thread_lock = threading.Lock() engine.memory_lock = threading.RLock() engine.running = False + engine.run_generation = 0 engine.project = {"nodes": {"start": {"type": "start"}}} engine.paused = False engine.stop_event = Event() From 0571001636a9d4babf889d635d424f246650068d Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:10:39 +0800 Subject: [PATCH 09/10] Protect shared execution statistics during parallel node updates --- main.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/main.py b/main.py index f74a900..aeb261c 100644 --- a/main.py +++ b/main.py @@ -558,7 +558,7 @@ def clear(self): class AutomationCore: def __init__(self, log_callback, app_instance): self.running = False; self.paused = False; self.stop_event = threading.Event(); self.pause_event = threading.Event() - self.log = log_callback; self.app = app_instance; self.project = None; self.runtime_memory = {}; self.memory_lock = threading.RLock(); self.io_lock = threading.Lock(); self.run_generation = 0 + self.log = log_callback; self.app = app_instance; self.project = None; self.runtime_memory = {}; self.memory_lock = threading.RLock(); self.stats_lock = threading.Lock(); self.io_lock = threading.Lock(); self.run_generation = 0 self.active_threads = 0; self.thread_lock = threading.Lock(); self.scaling_ratio = 1.0; self.breakpoints = set() self.max_threads = 50 self.pending_nodes = [] @@ -694,7 +694,7 @@ def _run_flow_engine(self, start_node_id=None, generation=None): if generation is None: generation = self.run_generation current_run = generation == self.run_generation - stats = dict(self.performance_stats) if current_run else None + stats = self._stats_snapshot() if current_run else None if current_run: self.running = False if stats is not None: @@ -738,11 +738,11 @@ def _process_node_thread(self, node_id): self.app.highlight_node_safe(node_id, 'running'); self.app.select_node_safe(node_id) try: out_port = self._execute_node(node) - self.performance_stats['nodes_executed'] += 1 + self._increment_stat('nodes_executed') except Exception as e: self.log(f"💥 节点[{node_id}]错误: {e}", "error"); traceback.print_exc(); - self.performance_stats['errors'] += 1 + self._increment_stat('errors') out_port = 'fail' if out_port == '__STOP__' or self.stop_event.is_set(): return @@ -778,6 +778,14 @@ def _memory_snapshot(self): with self.memory_lock: return dict(self.runtime_memory) + def _increment_stat(self, key): + with self.stats_lock: + self.performance_stats[key] += 1 + + def _stats_snapshot(self): + with self.stats_lock: + return dict(self.performance_stats) + def _replace_variables(self, text): if not isinstance(text, str): return str(text) for k, v in self._memory_snapshot().items(): From 26e3d8494350a3de9135615416b1753343c8a861 Mon Sep 17 00:00:00 2001 From: QwejayHuang Date: Wed, 14 Oct 2026 00:10:53 +0800 Subject: [PATCH 10/10] Stress-test execution statistics under concurrent updates --- tests/test_regressions.py | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/tests/test_regressions.py b/tests/test_regressions.py index a7532f7..d6eb280 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -75,6 +75,10 @@ def setUpClass(cls): "AutomationCore", "_memory_snapshot", {"threading": threading} ) + cls.IncrementStatProbe = extract_class_with_method( + "AutomationCore", "_increment_stat", + {"threading": threading} + ) cls.FinishUiProbe = extract_class_with_method( "AutomationCore", "_finish_run_ui", {"threading": threading} @@ -200,6 +204,30 @@ def after(self, delay, callback): self.assertTrue(engine.pause_event.is_set()) self.assertEqual(engine.pending_nodes, []) + def test_parallel_stat_updates_do_not_lose_counts(self): + import threading + + core = self.IncrementStatProbe() + core.stats_lock = threading.Lock() + core.performance_stats = {"nodes_executed": 0} + workers = 8 + increments = 5000 + threads = [ + threading.Thread( + target=lambda: [ + core._increment_stat("nodes_executed") + for _ in range(increments) + ] + ) + for _ in range(workers) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + self.assertEqual(core.performance_stats["nodes_executed"], workers * increments) + def test_stale_run_cannot_reset_new_run_ui(self): import threading