# G1.5 regression stress driver: N iterations of LOAD/OPEN_GUI/PLAY/STOP/WM_CLOSE # on Nexus + EZkeys. Watch: bridge process alive, no new EXCEPTION/FAILED, writeIndex advancing. import ctypes, ctypes.wintypes as w, struct, subprocess, sys, os, time, re sys.stdout.reconfigure(encoding='utf-8', errors='replace') APP_PID = int(sys.argv[1]) ITERS = int(sys.argv[2]) if len(sys.argv) > 2 else 5 SHM_NAME = f"SonicForge_DAW_IPC_{APP_PID}".encode() APP_LOG = os.path.join(os.environ.get('APPDATA', '.'), 'SonicForgeDAW', 'logs', 'bridge.log') NEXUS = r"C:\Program Files\Common Files\VST3\Nexus.vst3" EZKEYS = r"C:\Program Files\Common Files\VST3\Toontrack\EZkeys 2.vst3" kernel32 = ctypes.WinDLL('kernel32', use_last_error=True) kernel32.OpenFileMappingA.argtypes = [ctypes.wintypes.DWORD, w.BOOL, ctypes.c_char_p] kernel32.OpenFileMappingA.restype = ctypes.c_void_p kernel32.MapViewOfFile.argtypes = [ctypes.c_void_p, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.c_size_t] kernel32.MapViewOfFile.restype = ctypes.c_void_p hMap = kernel32.OpenFileMappingA(0xF001F, False, SHM_NAME) if not hMap: print(f"OpenFileMapping failed err={ctypes.get_last_error()}"); sys.exit(1) ptr = kernel32.MapViewOfFile(hMap, 0xF001F, 0, 0, 0) print(f"SHM mapped ({SHM_NAME}) iters={ITERS}") def w32(off, fmt, *args): ctypes.memmove(ptr + off, struct.pack(fmt, *args), struct.calcsize(fmt)) def rd32(off): return ctypes.c_uint32.from_address(ptr + off).value def write_control(ctype, arg0, arg1, ch, path): for i in range(8): off = 2836 + i * 1040 if rd32(off) == 0: w32(off, 'IIII', ctype, arg0, arg1, ch) if path: b = path.encode('utf-8')[:1023] + b'\x00' ctypes.memmove(ptr + off + 16, b, len(b)) return i raise RuntimeError("control queue full") def push_controls(events): for i in range(8): w32(2836 + i * 1040, 'I', 0) for (ctype, arg0, arg1, ch, path) in events: write_control(ctype, arg0, arg1, ch, path) w32(11156, 'I', len(events)) def log_lines(): try: with open(APP_LOG, 'r', encoding='utf-8', errors='replace') as f: return f.readlines() except FileNotFoundError: return [] def wait_log(pattern, timeout, since): end = time.time() + timeout while time.time() < end: for i in range(since, len(log_lines())): if pattern in log_lines()[i]: return log_lines()[i] time.sleep(0.2) return None def wait_gui(ch, timeout, since): end = time.time() + timeout while time.time() < end: for i in range(since, len(log_lines())): l = log_lines()[i] if "GUI attached hwnd=" in l and f"ch={ch}" in l: return l time.sleep(0.2) return None def bridge_pid(): out = subprocess.run(['powershell', '-Command', '(Get-Process daw_vst_bridge -ErrorAction SilentlyContinue).Id'], capture_output=True, text=True).stdout.strip().splitlines() return [int(x) for x in out if x.strip().isdigit()] def bad_since(since): return [l for l in log_lines()[since:] if 'EXCEPTION' in l or 'FAILED' in l] def find_hwnd(ch): last = None for l in log_lines(): if "GUI attached hwnd=" in l and f"ch={ch}" in l: m = re.search(r'hwnd=([0-9A-Fa-f]+)', l) if m: last = int(m.group(1), 16) return last fails = 0 for it in range(ITERS): print(f"--- iter {it} ---") since = len(log_lines()) # LOAD Nexus ch5 (already loaded after G2.4 guard -> "LOAD skip" also OK) push_controls([(2, 0, 0, 5, NEXUS)]) if not (wait_log("instrument loaded ch=5 type=0", 30, since) or wait_log("LOAD skip ch=5", 30, since)): print(f"iter{it} LOAD Nexus TIMEOUT"); fails += 1; continue # OPEN_GUI ch5 push_controls([(4, 0, 0, 5, NEXUS)]) if not wait_gui(5, 30, since): print(f"iter{it} OPEN_GUI Nexus TIMEOUT"); fails += 1; continue # PLAY -> writeIndex advancing while editor open (G1.4 core) wi0 = rd32(4) push_controls([(3, 1, 0, 0, "")]) if not wait_log("transport PLAY", 10, since): print(f"iter{it} PLAY TIMEOUT"); fails += 1; continue time.sleep(1.5) wi1 = rd32(4) ok = (wi1 - wi0) > 50 print(f"iter{it} PLAY editor-open writeIndex {wi0}->{wi1} delta={wi1-wi0} {'OK' if ok else 'STALLED'}") if not ok: fails += 1 # STOP push_controls([(3, 0, 0, 0, "")]) if not wait_log("transport STOP", 10, since): print(f"iter{it} STOP TIMEOUT"); fails += 1; continue # WM_CLOSE ch5 -> state restored hwnd = find_hwnd(5) if hwnd: ctypes.windll.user32.PostMessageW(hwnd, 0x0010, 0, 0) r = wait_log("state restored ch=5", 30, since) print(f"iter{it} WM_CLOSE Nexus hwnd={hex(hwnd)}: {'OK' if r else 'TIMEOUT'}") if not r: fails += 1 else: print(f"iter{it} hwnd Nexus not found"); fails += 1 # LOAD EZkeys ch5 + OPEN_GUI + PLAY + STOP + CLOSE since = len(log_lines()) push_controls([(2, 0, 0, 5, EZKEYS)]) if not wait_log("instrument loaded ch=5 type=0", 30, since): print(f"iter{it} LOAD EZkeys TIMEOUT"); fails += 1; continue push_controls([(4, 0, 0, 5, EZKEYS)]) if not wait_gui(5, 30, since): print(f"iter{it} OPEN_GUI EZkeys TIMEOUT"); fails += 1; continue push_controls([(3, 1, 0, 0, "")]) wait_log("transport PLAY", 10, since) time.sleep(1.5) push_controls([(3, 0, 0, 0, "")]) wait_log("transport STOP", 10, since) hwnd = find_hwnd(5) if hwnd: ctypes.windll.user32.PostMessageW(hwnd, 0x0010, 0, 0) r = wait_log("state restored ch=5", 30, since) print(f"iter{it} WM_CLOSE EZkeys hwnd={hex(hwnd)}: {'OK' if r else 'TIMEOUT'}") if not r: fails += 1 else: print(f"iter{it} hwnd EZkeys not found"); fails += 1 bad = bad_since(since) if bad: print(f"iter{it} EXCEPTION/FAILED: {bad[-2:]}"); fails += 1 # final watch: bridge alive + no new bad since = len(log_lines()) time.sleep(3) bad = bad_since(since) alive = bridge_pid() print(f"bridge alive: {alive} (expect non-empty)") if alive else print("BRIDGE DEAD") if not alive: fails += 1 print(f"G1.5 RESULT: {'PASS' if fails == 0 else f'FAIL ({fails} issues)'}") sys.exit(0 if fails == 0 else 1)