fix: clean up stale BullMQ/Redis data and improve process termination in bbdev

Author: shirohasuki

api/steps/sardine/01_run_event_step.py

@@ -29,7 +29,7 @@ async def handler(data, context):
 
     sardine_dir = f"{bbdir}/bb-tests/sardine"
 
-    command = f"python3 run_tests.py --allure -m \"({data.get('workload', '')})\""
+    command = f"python3 run_tests.py -m \"({data.get('workload', '')})\""
     context.logger.info(
         "Executing sardine command", {"command": command, "cwd": sardine_dir}
     )

api/steps/verilator/04_sim_event_step.py

@@ -40,6 +40,7 @@ async def handler(data, context):
     )
 
     binary_path = search_workload(f"{bbdir}/bb-tests/output/workloads/src", binary_name)
+    context.logger.info(f"binary_path: {binary_path}")    
     success_result, failure_result = await check_result(
         context, returncode=(binary_path == None), continue_run=True
     )

bbdev

@@ -4,9 +4,11 @@ import sys
 import argparse
 import subprocess
 import os
+import signal
 import shlex
 import json
 import time
+import shutil
 import requests
 
 from utils import find_available_port
@@ -441,6 +443,11 @@ if __name__ == "__main__":
             "uvm",
             "palladium",
         ]:
+            # 0. Clean up stale BullMQ/Redis data to prevent replaying old events
+            aof_dir = os.path.join(workflow_dir, ".motia", "appendonlydir")
+            if os.path.exists(aof_dir):
+                shutil.rmtree(aof_dir)
+
             # 1. Start service in background ================================
             # If port is specified, use the specified port; otherwise, automatically assign port
             if args.port:
@@ -451,7 +458,8 @@ if __name__ == "__main__":
             proc = subprocess.Popen(
                 ["pnpm", "dev", "--port", str(available_port)],
                 cwd=workflow_dir,
-                stderr=subprocess.DEVNULL  # Suppress stderr to hide BullMQ errors
+                stderr=subprocess.DEVNULL,  # Suppress stderr to hide BullMQ errors
+                preexec_fn=os.setsid  # Create new process group
             )
 
             # Wait for service to start
@@ -479,7 +487,11 @@ if __name__ == "__main__":
                     check=False,
                     text=True,
                 )
-                proc.terminate()
+                # Kill entire process group to clean up all child processes
+                try:
+                    os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
+                except ProcessLookupError:
+                    pass
                 proc.wait()
                 sys.exit(1)
 
@@ -510,7 +522,8 @@ if __name__ == "__main__":
             # 3. Shutdown service ================================
             # Give observability plugin time to finish async operations (e.g., Redis writes)
             time.sleep(1)
-            proc.terminate()
+            # Kill entire process group (pnpm -> motia -> node) to avoid orphan processes
+            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
             proc.wait()
             print(
                 f"\nTask completed. Command running on http://localhost:{available_port} is finished"