feat(companion): Close manager-only fds in the companion child

spawn_process now closes the manager's control socket listener and wakeup
self-pipe in the forked companion before running its target. Both are
inherited across the fork; closing them stops a companion from holding the
control listener (and possibly answering control requests) or the manager's
private signal pipe. Guarded so direct spawns without a control socket or
running loop are a no-op.

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

docs/design/companion-process-manager.md

@@ -692,7 +692,7 @@ No per-companion logic in Arbiter.
 - [x] Add manager shutdown handling in Arbiter.
 - [x] Wire Gunicorn reload to manager `reread` or restart.
 - [x] Close Gunicorn-only fds in manager child.
-- [ ] Close manager-only fds in companion child.
+- [x] Close manager-only fds in companion child.
 - [ ] Add parent-death cleanup.
 - [ ] Add lifecycle logs.
 - [ ] Add tests for config validation.

gunicorn/companion/manager.py

@@ -282,6 +282,7 @@ class CompanionManager:
             return pid
 
         try:
+            self._close_manager_fds()
             self._apply_environment(process.config)
             self._redirect_output(process.config)
             target = self._resolve_target(process.config.target)
@@ -293,6 +294,23 @@ class CompanionManager:
             os._exit(1)
         os._exit(0)
 
+    def _close_manager_fds(self) -> None:
+        """Close the manager's own fds in a freshly forked companion.
+
+        A companion inherits the manager's control socket and wakeup pipe but
+        must not keep them: an open listener would let a companion answer
+        control requests, and the pipe is the manager's private signal path.
+        Both are closed before the target runs.
+        """
+        if self.control is not None and self.control.listener is not None:
+            self.control.listener.close()
+        if self._wakeup_pipe is not None:
+            for fd in self._wakeup_pipe:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+
     def start_process(self, name: str):
         """Start a companion by name (the control ``start`` command).
 

tests/test_companion_manager.py

@@ -40,6 +40,21 @@ def test_resolve_target_rejects_bad_string():
         CompanionManager._resolve_target("no_colon")
 
 
+def test_close_manager_fds_closes_control_and_pipe():
+    manager = make_manager("rq")
+    manager.control = mock.Mock()
+    manager._wakeup_pipe = (7, 8)
+    with mock.patch("os.close") as os_close:
+        manager._close_manager_fds()
+    manager.control.listener.close.assert_called_once_with()
+    assert os_close.call_count == 2
+
+
+def test_close_manager_fds_noop_when_unset():
+    manager = make_manager("rq")
+    manager._close_manager_fds()  # control and pipe are None: must not raise
+
+
 def test_apply_environment_sets_cwd_and_env():
     config = CompanionConfig(name="rq", target=lambda: None,
                              cwd="/tmp", env={"COMPANION_X": "1"})