fix: sync verify_output_token flaky fix from lab1 (retry loop with 5s deadline)

Author: andylizf

harness/verifier.py

@@ -33,6 +33,7 @@ def verify_env_check(
         expect_rows: Optional[int] = None,
         expect_cols: Optional[int] = None,
     ) -> Result:
+        """Verify that probe reported correct environment."""
         msgs = self._sb.get_messages(session_id=session_id, msg_type="env_check")
         if not msgs:
             return Result(False, f"No env_check message from session {session_id}")
@@ -52,6 +53,7 @@ def verify_env_check(
         return Result(True, "env_check OK", data=msg)
 
     def verify_pgrp_isolation(self, sessions: list[str]) -> Result:
+        """Verify that probes in different panes have different process groups."""
         pgrps = {}
         for sid in sessions:
             msgs = self._sb.get_messages(session_id=sid, msg_type="env_check")
@@ -68,6 +70,7 @@ def verify_signal_delivery(
         session_id: str,
         signal_name: str,
     ) -> Result:
+        """Verify that probe received a specific signal."""
         msgs = self._sb.get_messages(session_id=session_id, msg_type="signal")
         matching = [m for m in msgs if m.get("signal") == signal_name]
         if not matching:
@@ -84,9 +87,13 @@ def verify_signal_isolation(
         inactive: list[str],
         signal: str,
     ) -> Result:
+        """Verify signal was received by active pane but NOT by inactive panes."""
+        # Active should have it
         active_result = self.verify_signal_delivery(active, signal)
         if not active_result.passed:
             return Result(False, f"Active pane {active} did not receive {signal}")
+
+        # Inactive should not
         for sid in inactive:
             msgs = self._sb.get_messages(session_id=sid, msg_type="signal")
             matching = [m for m in msgs if m.get("signal") == signal]
@@ -99,20 +106,53 @@ def verify_signal_isolation(
         return Result(True, f"{signal} isolation OK: {active} got it, {inactive} did not")
 
     def verify_output_token(self, session_id: str) -> Result:
-        msgs = self._sb.get_messages(session_id=session_id, msg_type="output_token")
-        if not msgs:
-            return Result(False, f"No output_token from {session_id}")
-        token = msgs[0]["token"]
-        raw = self._driver.read_raw(timeout=2.0)
-        if token.encode() in raw:
-            return Result(True, "output_token passthrough OK", data={"token": token})
+        """Verify that probe's output token appeared in client byte stream.
+
+        1. Wait for probe to emit a fresh token via sideband (proves probe alive).
+        2. Read client output and check if that token traversed the PTY pipeline.
+
+        Probe sends sideband first then stdout, so by the time we see the
+        sideband message, the token is already in the PTY/client buffer.
+        """
+        import time as _time
+        old_count = len(self._sb.get_messages(
+            session_id=session_id, msg_type="output_token"))
+        # Step 1: wait for a fresh sideband token (probe emits every ~1s)
+        deadline = _time.monotonic() + 3.0
+        new_msgs = []
+        while _time.monotonic() < deadline:
+            new_msgs = self._sb.get_messages(
+                session_id=session_id, msg_type="output_token")[old_count:]
+            if new_msgs:
+                break
+            _time.sleep(0.2)
+        if not new_msgs:
+            return Result(False,
+                         f"No new output_token from probe within 3s "
+                         f"(had {old_count} before)")
+        # Step 2: read client output in a loop — under high-throughput
+        # scenarios, read_raw may return early (pexpect shrinks timeout
+        # after the first chunk), so retry until the token appears or we
+        # hit the overall deadline.
+        raw = b""
+        read_deadline = _time.monotonic() + 5.0
+        while _time.monotonic() < read_deadline:
+            raw += self._driver.read_raw(timeout=1.0)
+            for m in reversed(new_msgs):
+                if m["token"].encode() in raw:
+                    return Result(True, "output_token passthrough OK",
+                                 data={"token": m["token"]})
+            _time.sleep(0.1)
         return Result(
             False,
-            f"Token {token} not found in client output ({len(raw)} bytes read)",
-            data={"token": token, "raw_len": len(raw)},
+            f"Token {new_msgs[-1]['token']} not found in client output "
+            f"({len(raw)} bytes read, {len(new_msgs)} new tokens from sideband)",
+            data={"token": new_msgs[-1]["token"], "raw_len": len(raw),
+                  "new_token_count": len(new_msgs)},
         )
 
     def verify_input_token(self, session_id: str) -> Result:
+        """Send a token via client stdin and verify probe received it."""
         token = secrets.token_hex(16)
         self._driver.send_line(token)
 
@@ -128,3 +168,48 @@ def verify_input_token(self, session_id: str) -> Result:
             f"Probe did not report receiving token {token}",
             data={"sent": token, "received_msgs": msgs},
         )
+
+    def verify_file_contains(self, file_path: str, pattern: str) -> Result:
+        """Check that a file exists and contains the given pattern."""
+        if not os.path.exists(file_path):
+            return Result(False, f"File not found: {file_path}")
+        with open(file_path, "r", errors="replace") as f:
+            content = f.read()
+        if pattern in content:
+            return Result(True, f"Pattern found in {file_path}")
+        return Result(False, f"Pattern '{pattern}' not in {file_path} ({len(content)} bytes)")
+
+    def verify_fd_count(self, pid: int, max_fds: int) -> Result:
+        """Check that a process has at most max_fds open file descriptors."""
+        try:
+            proc = psutil.Process(pid)
+            fds = proc.num_fds()
+        except (psutil.NoSuchProcess, psutil.AccessDenied) as e:
+            return Result(False, f"Cannot check fds: {e}")
+        passed = fds <= max_fds
+        return Result(passed, f"fd count: {fds} (max {max_fds})", data={"fds": fds})
+
+    def verify_winsize_update(
+        self,
+        session_id: str,
+        rows: int,
+        cols: int,
+    ) -> Result:
+        """After resize, verify probe got SIGWINCH and updated winsize."""
+        # Check for SIGWINCH
+        sig_result = self.verify_signal_delivery(session_id, "SIGWINCH")
+        if not sig_result.passed:
+            return Result(False, "No SIGWINCH received after resize")
+        # Check latest env_check for updated winsize
+        msgs = self._sb.get_messages(session_id=session_id, msg_type="env_check")
+        if not msgs:
+            return Result(False, "No env_check after SIGWINCH")
+        latest = msgs[-1]
+        ws = latest.get("winsize", {})
+        if ws.get("rows") == rows and ws.get("cols") == cols:
+            return Result(True, f"winsize updated to {rows}x{cols}", data=latest)
+        return Result(
+            False,
+            f"Expected {rows}x{cols}, got {ws.get('rows')}x{ws.get('cols')}",
+            data=latest,
+        )