feat(companion): Restart the manager on Gunicorn reload

Arbiter.reload (SIGHUP) now calls reload_companion_manager. A running manager
is sent SIGTERM so it drains its companions; the SIGCHLD reaper clears its pid
and manage_companion_manager respawns it from the freshly reloaded cfg. If
companions were added where none ran, a new manager starts immediately.

Restarting reuses the existing stop and respawn path; transactional
per-companion reread stays available separately through the control socket.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

docs/design/companion-process-manager.md

@@ -690,7 +690,7 @@ No per-companion logic in Arbiter.
 - [x] Implement transactional `reread`.
 - [x] Add manager spawn/reap logic in Arbiter.
 - [x] Add manager shutdown handling in Arbiter.
-- [ ] Wire Gunicorn reload to manager `reread` or restart.
+- [x] Wire Gunicorn reload to manager `reread` or restart.
 - [ ] Close Gunicorn-only fds in manager child.
 - [ ] Close manager-only fds in companion child.
 - [ ] Add parent-death cleanup.

gunicorn/arbiter.py

@@ -495,6 +495,9 @@ class Arbiter:
         # manage workers
         self.manage_workers()
 
+        # reload companions with the new configuration
+        self.reload_companion_manager()
+
     def murder_workers(self):
         """\
         Kill unused/idle workers
@@ -705,6 +708,21 @@ class Arbiter:
             self.log.exception("Exception in companion manager process")
             sys.exit(-1)
 
+    def reload_companion_manager(self):
+        """Restart the companion manager so it picks up the new configuration.
+
+        Gunicorn reload (SIGHUP) rebuilds cfg. A running manager is asked to
+        stop -- it drains its companions first -- and the SIGCHLD reaper then
+        clears its pid so manage_companion_manager respawns it from the fresh
+        cfg. If companions were added where none ran, a new manager starts
+        right away. Per-companion transactional reread stays available
+        separately through the control socket.
+        """
+        if self.companion_manager_pid != 0:
+            self.log.info("Reloading companion manager")
+            self.stop_companion_manager(signal.SIGTERM)
+        self.manage_companion_manager()
+
     def stop_companion_manager(self, sig):
         """Signal the companion manager to exit, if it is running.
 

tests/test_arbiter.py

@@ -205,6 +205,28 @@ def test_stop_companion_manager_clears_pid_when_already_gone():
     assert arbiter.companion_manager_pid == 0
 
 
+def test_reload_companion_manager_restarts_running():
+    arbiter = gunicorn.arbiter.Arbiter(DummyApplication())
+    arbiter.cfg.set("companion_workers", [{"name": "rq", "target": "pkg:run"}])
+    arbiter.companion_manager_pid = 4242
+    arbiter.stop_companion_manager = mock.Mock()
+    arbiter.spawn_companion_manager = mock.Mock()
+    arbiter.reload_companion_manager()
+    arbiter.stop_companion_manager.assert_called_once_with(signal.SIGTERM)
+    # pid still set (stop is mocked), so no respawn until the old one is reaped
+    arbiter.spawn_companion_manager.assert_not_called()
+
+
+def test_reload_companion_manager_starts_when_none_running():
+    arbiter = gunicorn.arbiter.Arbiter(DummyApplication())
+    arbiter.cfg.set("companion_workers", [{"name": "rq", "target": "pkg:run"}])
+    arbiter.stop_companion_manager = mock.Mock()
+    arbiter.spawn_companion_manager = mock.Mock()
+    arbiter.reload_companion_manager()
+    arbiter.stop_companion_manager.assert_not_called()
+    arbiter.spawn_companion_manager.assert_called_once_with()
+
+
 @mock.patch('gunicorn.sock.close_sockets')
 def test_arbiter_stop_signals_companion_manager(close_sockets):
     arbiter = gunicorn.arbiter.Arbiter(DummyApplication())