# G1.2 verification against live app: single UiThread. # Flow: PLAY(audio advancing) -> OPEN_GUI ch5 during PLAY (deferred, audio still alive) # -> STOP (attach ch5) -> PLAY again -> OPEN_GUI ch6 (Option B closes ch5 editor # synchronously, ch6 deferred) -> STOP (attach ch6) -> close ch6 editor via WM_CLOSE # -> no EXCEPTION/FAILED/restart anywhere, writeIndex advancing while PLAY. import ctypes, ctypes.wintypes as w, struct, subprocess, sys, os, time sys.stdout.reconfigure(encoding='utf-8', errors='replace') APP_PID = int(sys.argv[1]) 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" 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})") 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, verbose=True): end = time.time() + timeout while time.time() < end: lines = log_lines() for i in range(since, len(lines)): if pattern in lines[i]: if verbose: print(f" LOG: {lines[i].strip()}") return lines[i].strip() time.sleep(0.2) return None def write_index(): return rd32(4) def sample_wi(seconds): a = write_index(); time.sleep(seconds); b = write_index() return a, b def bridge_pid(): try: out = subprocess.run(["cmd", "/c", "tasklist /FI \"IMAGENAME eq daw_vst_bridge.exe\" /FO CSV"], capture_output=True, text=True).stdout for line in out.splitlines()[1:]: if "daw_vst_bridge" in line: return line.split(',')[1].strip('"') except Exception: pass return None bad = [] def collect_bad(since): for l in log_lines()[since:]: if any(k in l for k in ("EXCEPTION", "FAILED", "restarted")): bad.append(l.strip()) since = len(log_lines()) pid0 = bridge_pid() print("bridge pid:", pid0) # --- Phase A: PLAY, audio advancing --- push_controls([(3, 1, 0, 0, "")]) r = wait_log("transport PLAY", 10, since) print("A transport:", r or "TIMEOUT") a, b = sample_wi(3) print(f"A writeIndex {a}->{b} delta={b-a} ({'ADVANCING' if b > a else 'STALLED'})") if b <= a: bad.append("A: writeIndex stalled during PLAY") # --- Phase B: OPEN_GUI ch5 during PLAY (expect deferred) --- since = len(log_lines()) push_controls([(4, 0, 0, 5, "Nexus")]) r = wait_log("transport PLAY) - queued", 10, since) print("B defer ch5:", r or "TIMEOUT") a, b = sample_wi(3) print(f"B writeIndex {a}->{b} delta={b-a} ({'ADVANCING' if b > a else 'STALLED'})") if b <= a: bad.append("B: writeIndex stalled while GUI deferred") collect_bad(since) # --- Phase C: STOP -> attach ch5 --- since = len(log_lines()) push_controls([(3, 0, 0, 0, "")]) r = wait_log("transport STOP", 10, since) print("C transport:", r or "TIMEOUT") r = wait_log("GUI attached hwnd=", 30, since) print("C attach ch5:", r or "TIMEOUT") # --- Phase D: PLAY again -> OPEN_GUI ch6 (Option B synchronous close of ch5) --- since = len(log_lines()) push_controls([(3, 1, 0, 0, "")]) r = wait_log("transport PLAY", 10, since) print("D transport:", r or "TIMEOUT") a, b = sample_wi(2) print(f"D writeIndex {a}->{b} delta={b-a} ({'ADVANCING' if b > a else 'STALLED'})") if b <= a: bad.append("D: writeIndex stalled during PLAY #2") since = len(log_lines()) push_controls([(4, 0, 0, 6, "Nexus")]) r = wait_log("transport PLAY) - queued", 10, since) print("D defer ch6:", r or "TIMEOUT (maybe attach directly?)") time.sleep(1) collect_bad(since) # ch5 editor must be closed (Option B) — check log for close editor of ch5 closehit = wait_log("closeGUI", 15, since, verbose=False) or "" print("D close ch5 editor:", closehit or "no closeGUI log (check)") # --- Phase E: STOP -> attach ch6 --- since = len(log_lines()) push_controls([(3, 0, 0, 0, "")]) r = wait_log("transport STOP", 10, since) print("E transport:", r or "TIMEOUT") r = wait_log("GUI attached hwnd=", 30, since) print("E attach ch6:", r or "TIMEOUT") collect_bad(since) # --- Phase F: close ch6 editor via WM_CLOSE (post_close_gui path) --- since = len(log_lines()) # find ch6 hwnd from log hwnd6 = None for l in log_lines(): if "ch=6" in l and "GUI attached" in l: import re m = re.search(r'hwnd=([0-9A-Fa-f]+)', l) if m: hwnd6 = int(m.group(1), 16) if hwnd6: ctypes.windll.user32.PostMessageW(hwnd6, 0x0010, 0, 0) # WM_CLOSE print("F posted WM_CLOSE to ch6 hwnd", hex(hwnd6)) else: print("F hwnd6 not found") r = wait_log("closeGUI reload", 30, since) print("F close ch6:", r or "TIMEOUT") collect_bad(since) # --- watch 8s for crash/restart --- since = len(log_lines()) end = time.time() + 8 while time.time() < end: collect_bad(since) pid = bridge_pid() if pid != pid0: bad.append(f"BRIDGE RESTART {pid0}->{pid}") break time.sleep(0.2) print() if bad: print("G1.2 RESULT: FAIL") for x in bad[:20]: print(" -", x) else: print("G1.2 RESULT: PASS - 1 UiThread, defer OK, Option B OK, close OK, audio advancing, no crash")