Drive the acknowledgment window from a hook instead of a parameter

spawn() no longer carries pre_ack_delay through the handshake for the tests'
benefit. An empty _pre_ack_hook() sits where the delay ran, and the cases that
need the window open override it on a subclass, as they already do for the
select and master-descriptor seams.

Author: gorandodig

render_machine/_posix_pty.py

@@ -501,14 +501,8 @@ def spawn(
         stop_event: Optional[threading.Event] = None,
         input_driver: Optional[object] = None,
         handshake_timeout: float = HANDSHAKE_TIMEOUT_SECONDS,
-        pre_ack_delay: float = 0.0,
     ) -> None:
-        """Allocates the terminal, launches the target, and returns once it is running.
-
-        `pre_ack_delay` holds the parent's acknowledgment for a bounded time. It exists so
-        the barrier's window can be driven deterministically from tests; production
-        callers leave it at zero.
-        """
+        """Allocates the terminal, launches the target, and returns once it is running."""
         if self._spawned:
             raise RuntimeError("PosixPtyProcess instances are single-use")
         self._spawned = True
@@ -521,7 +515,7 @@ def spawn(
             self._open_channels()
             self._start_child(command, cwd, env)
             self._hand_over_to_reader()
-            self._run_handshake(deadline, pre_ack_delay)
+            self._run_handshake(deadline)
             self._close_owned("_status_r")  # the handshake has resolved
         except BaseException:
             self._rollback()
@@ -715,7 +709,7 @@ def _hand_over_to_reader(self) -> None:
 
     # ---------------------------------------------------------------- handshake
 
-    def _run_handshake(self, deadline: float, pre_ack_delay: float) -> None:
+    def _run_handshake(self, deadline: float) -> None:
         parser = _HandshakeParser()
         assert self._proc is not None and self._proc.stderr is not None
         status_r, err_r = self._status_r, self._err_r
@@ -733,12 +727,10 @@ def _run_handshake(self, deadline: float, pre_ack_delay: float) -> None:
                 watched.discard(stderr_fd)
             if err_r in readable:
                 self._consume_reader_edge(watched, err_r)
-            if status_r in readable and self._advance_handshake(parser, status_r, deadline, pre_ack_delay):
+            if status_r in readable and self._advance_handshake(parser, status_r, deadline):
                 return
 
-    def _advance_handshake(
-        self, parser: _HandshakeParser, status_r: int, deadline: float, pre_ack_delay: float
-    ) -> bool:
+    def _advance_handshake(self, parser: _HandshakeParser, status_r: int, deadline: float) -> bool:
         """Feeds one status chunk. Returns True once exec has been observed."""
         chunk = os.read(status_r, READ_CHUNK_BYTES)
         try:
@@ -752,17 +744,16 @@ def _advance_handshake(
             reason = parser.failure_payload.decode("utf-8", "replace")
             raise TerminalLaunchError(self._launch_message(f"the launcher failed: {reason}"))
         if parser.session_ready and not self._acked:
-            self._acknowledge(deadline, pre_ack_delay)
+            self._acknowledge(deadline)
         return False
 
-    def _acknowledge(self, deadline: float, pre_ack_delay: float) -> None:
+    def _acknowledge(self, deadline: float) -> None:
         """Records the group, delivers the no-driver VEOF, and only then releases the target."""
         assert self._proc is not None
         self._pgid = self._proc.pid  # recorded BEFORE the target can run
         if self._input_driver is None:
             self._inject_veof(deadline)
-        if pre_ack_delay > 0:
-            self._wait_pre_ack(pre_ack_delay, deadline)
+        self._pre_ack_hook()
         self._acked = True
         ack_w = self._ack_w
         assert ack_w is not None
@@ -774,12 +765,9 @@ def _acknowledge(self, deadline: float, pre_ack_delay: float) -> None:
             console.debug("the launcher closed the acknowledgment pipe before the parent acknowledged")
         self._close_owned("_ack_w")
 
-    def _wait_pre_ack(self, delay: float, deadline: float) -> None:
-        until = min(time.monotonic() + delay, deadline)
-        while time.monotonic() < until:
-            self._check_cancelled()
-            self._check_reader_failed()
-            time.sleep(min(POLL_INTERVAL_SECONDS, max(0.0, until - time.monotonic())))
+    def _pre_ack_hook(self) -> None:
+        """The window between the recorded group and the acknowledgment that releases the
+        target. Empty in production; a test overrides it to hold the window open."""
 
     def _inject_veof(self, deadline: float) -> None:
         result, receipt = self._input_queue.submit(

tests/test_terminal_process.py

@@ -773,31 +773,48 @@ def test_reader_exits_cleanly_when_the_leader_exits_with_a_descendant_on_the_sla
     assert process.reader_exc is None
 
 
+# How long the delayed-ack backend below holds the acknowledgment window open. Nothing
+# waits it out: every case that uses it ends the window itself.
+ACK_WINDOW_SECONDS = 5.0
+
+
+class _DelayedAckProcess(_posix_pty.PosixPtyProcess):
+    """Holds the acknowledgment for a bounded time, so the barrier's window is opened
+    rather than raced. The wait is cancellable and reader-aware, like the path it sits in."""
+
+    def __init__(self, delay=ACK_WINDOW_SECONDS):
+        super().__init__()
+        self.delay = delay
+        self.entered = threading.Event()
+
+    def _pre_ack_hook(self):
+        self.entered.set()
+        until = time.monotonic() + self.delay
+        while time.monotonic() < until:
+            self._check_cancelled()
+            self._check_reader_failed()
+            time.sleep(min(0.02, max(0.0, until - time.monotonic())))
+
+
 def test_cancellation_inside_the_ack_window_leaves_nothing_behind():
     """Deterministic through the delayed-ack hook: the window is opened, not raced.
 
     The cancellation waits for the hook to be entered, so it can never land before
     SESSION_READY however slowly the launcher gets there.
     """
     stop_event = threading.Event()
-    process = _posix_pty.PosixPtyProcess()
-    entered = threading.Event()
-    real_wait_pre_ack = process._wait_pre_ack
-
-    def recording_wait_pre_ack(delay, deadline):
-        entered.set()
-        real_wait_pre_ack(delay, deadline)
+    process = _DelayedAckProcess()
+    entered = process.entered
 
     def cancel_inside_the_window():
         if entered.wait(SPAWN_TIMEOUT):
             stop_event.set()
 
-    process._wait_pre_ack = recording_wait_pre_ack
     canceller = threading.Thread(target=cancel_inside_the_window, daemon=True)
     canceller.start()
     try:
         with pytest.raises(RenderCancelledError):
-            process.spawn(["/bin/sh", "-c", "sleep 30"], stop_event=stop_event, pre_ack_delay=5.0)
+            process.spawn(["/bin/sh", "-c", "sleep 30"], stop_event=stop_event)
     finally:
         canceller.join(timeout=SHORT_TIMEOUT)
         process.close()
@@ -855,10 +872,10 @@ def test_a_tree_that_exits_on_sigterm_does_not_wait_out_the_whole_grace(tmp_path
 def test_launcher_ack_timeout_beats_the_parents_ack():
     """The parent's write hits a closed pipe; the launcher's own reason must surface."""
     env = dict(os.environ, **{pty_exec.ACK_TIMEOUT_ENV: "0.2"})
-    process = _posix_pty.PosixPtyProcess()
+    process = _DelayedAckProcess(delay=2.0)
     try:
         with pytest.raises(TerminalLaunchError) as failure:
-            process.spawn(["/bin/sh", "-c", "exit 0"], env=env, pre_ack_delay=2.0, handshake_timeout=10.0)
+            process.spawn(["/bin/sh", "-c", "exit 0"], env=env, handshake_timeout=10.0)
     finally:
         process.close()
 
@@ -894,10 +911,10 @@ def recording_killpg(pgid, sig):
     monkeypatch.setattr(_posix_pty.os, "killpg", recording_killpg)
     stop_event = threading.Event()
     threading.Timer(0.2, stop_event.set).start()
-    process = _posix_pty.PosixPtyProcess()
+    process = _DelayedAckProcess()
     try:
         with pytest.raises(RenderCancelledError):
-            process.spawn(["/bin/sh", "-c", "sleep 30"], stop_event=stop_event, pre_ack_delay=5.0)
+            process.spawn(["/bin/sh", "-c", "sleep 30"], stop_event=stop_event)
     finally:
         process.close()