fix: Reap JobRunner subprocesses before os._exit to prevent orphans and semaphore leaks

software/control/core/job_processing.py

@@ -2,6 +2,7 @@
 import multiprocessing
 import queue
 import os
+import threading
 import time
 import json
 from datetime import datetime
@@ -925,6 +926,7 @@ def shutdown(self, timeout_s=1.0):
         self._output_queue = None
         self._shutdown_event = None
         self._pending_count = None
+        self._ready_event = None
 
     def run(self):
         import logging
@@ -1034,3 +1036,38 @@ def run(self):
         log_memory("WORKER_SHUTDOWN", include_children=False)
         stop_worker_monitoring()
         self._log.info("Shutdown request received, exiting run.")
+
+
+def shutdown_all_job_runners(timeout_s: float = 5.0) -> None:
+    """Best-effort teardown of any still-alive JobRunner subprocesses.
+
+    Called before the application exits. main_hcs.py exits via os._exit(), which
+    skips multiprocessing's atexit cleanup, so daemon JobRunner children that the
+    per-acquisition (non-blocking) shutdown threads have not finished with would
+    otherwise survive as orphans and leak their queue/event semaphores.
+    """
+    runners = [p for p in multiprocessing.active_children() if isinstance(p, JobRunner)]
+    if not runners:
+        return
+    log = squid.logging.get_logger("shutdown_all_job_runners")
+    log.info(f"Shutting down {len(runners)} job runner subprocess(es) before exit...")
+
+    def safe_shutdown(runner):
+        try:
+            runner.shutdown(timeout_s=max(1.0, timeout_s - 1.0))
+        except Exception as e:
+            # A concurrent per-acquisition shutdown thread may already be tearing
+            # this runner down; termination below still guarantees the process dies.
+            log.debug(f"Job runner shutdown raced or failed: {e}")
+
+    threads = [threading.Thread(target=safe_shutdown, args=(r,), daemon=True) for r in runners]
+    for t in threads:
+        t.start()
+    deadline = time.time() + timeout_s
+    for t in threads:
+        t.join(timeout=max(0.1, deadline - time.time()))
+    for runner in runners:
+        if runner.is_alive():
+            log.warning(f"Job runner {runner.pid} still alive after {timeout_s}s; terminating.")
+            runner.terminate()
+            runner.join(timeout=1.0)

software/control/core/multi_point_worker.py

@@ -697,7 +697,9 @@ def time_left():
         # Using daemon threads is safe here because:
         # 1. All jobs are complete and results are already drained
         # 2. The subprocess termination is best-effort cleanup only
-        # 3. If app exits before threads complete, OS will terminate subprocesses anyway
+        # 3. If the app exits before these threads complete, main_hcs.py calls
+        #    shutdown_all_job_runners() before os._exit() (os._exit skips the
+        #    multiprocessing atexit hook that would normally reap daemon children)
         # 4. This prevents slow subprocess termination from blocking acquisition completion
         log = self._log  # Capture for closure
 

software/main_hcs.py

@@ -453,5 +453,15 @@ def launch_claude_code():
     except Exception as e:
         log.warning(f"Error during shutdown abort handling: {e}")
 
+    # os._exit() below skips multiprocessing's atexit hook, so daemon JobRunner
+    # children still alive (e.g. their non-blocking shutdown threads lost the
+    # race) would be orphaned and leak their queue/event semaphores.
+    try:
+        from control.core.job_processing import shutdown_all_job_runners
+
+        shutdown_all_job_runners(timeout_s=5.0)
+    except Exception as e:
+        log.warning(f"Error shutting down job runners: {e}")
+
     logging.shutdown()  # Flush log handlers before os._exit() bypasses Python cleanup
     os._exit(exit_code)

software/tests/control/core/test_job_runner_teardown.py

@@ -0,0 +1,49 @@
+"""Tests for shutdown_all_job_runners (exit-time reaping of JobRunner children).
+
+main_hcs.py exits via os._exit(), which skips multiprocessing's atexit hook, so
+any JobRunner subprocess still alive at that point (its shutdown event never
+set — e.g. the exit-time abort join timed out, or the worker died before
+_finish_jobs) would be orphaned and leak its queue/event semaphores.
+shutdown_all_job_runners() is called right before os._exit() to close that hole.
+"""
+
+import multiprocessing
+import time
+
+from control.core.job_processing import JobRunner, shutdown_all_job_runners
+
+
+def _wait_dead(runner, timeout_s=10.0):
+    deadline = time.time() + timeout_s
+    while time.time() < deadline:
+        if not runner.is_alive():
+            return True
+        time.sleep(0.1)
+    return not runner.is_alive()
+
+
+def test_reaps_runner_that_was_never_told_to_stop():
+    runner = JobRunner()
+    runner.start()
+    assert runner.wait_ready(timeout_s=15.0), "job runner subprocess never became ready"
+    assert runner.is_alive()
+
+    shutdown_all_job_runners(timeout_s=5.0)
+
+    assert _wait_dead(runner), "job runner still alive after shutdown_all_job_runners()"
+    assert not any(isinstance(p, JobRunner) for p in multiprocessing.active_children())
+
+
+def test_noop_when_no_runners():
+    shutdown_all_job_runners(timeout_s=1.0)
+
+
+def test_safe_after_runner_already_shut_down():
+    runner = JobRunner()
+    runner.start()
+    assert runner.wait_ready(timeout_s=15.0)
+    runner.shutdown(timeout_s=5.0)
+    assert _wait_dead(runner)
+
+    shutdown_all_job_runners(timeout_s=2.0)
+    assert not any(isinstance(p, JobRunner) for p in multiprocessing.active_children())