From c0810f68a792b827892a145d5980dc9ac776bbe5 Mon Sep 17 00:00:00 2001
From: Scott Trinh <scott@scotttrinh.com>
Date: Wed, 15 Apr 2026 15:38:37 -0400
Subject: [PATCH 1/2] Stabilize PTY example command sync

The sandbox PTY example could observe a prompt before the shell was
ready to reliably echo command output, which made the example flaky
in CI.

Add an explicit round-trip readiness check and wait for a completion
sentinel so the example verifies full PTY command delivery before
exiting.
---
 examples/sandbox_14_pty_test.py | 43 +++++++++++++++++++++++++++++++--
 1 file changed, 41 insertions(+), 2 deletions(-)

diff --git a/examples/sandbox_14_pty_test.py b/examples/sandbox_14_pty_test.py
index efefc355..4a569549 100644
--- a/examples/sandbox_14_pty_test.py
+++ b/examples/sandbox_14_pty_test.py
@@ -20,6 +20,8 @@
 load_dotenv()
 
 EXPECTED_OUTPUT = "PTY_OK"
+READY_OUTPUT = "PTY_READY"
+DONE_OUTPUT = "PTY_DONE"
 PROMPT_MARKER = "$ "
 
 
@@ -42,6 +44,19 @@ async def _collect() -> bytes:
     return await asyncio.wait_for(_collect(), timeout=timeout)
 
 
+async def run_and_collect(
+    session: AsyncPTYSession,
+    command: str,
+    marker: str,
+    *,
+    timeout: float = 30.0,
+) -> bytes:
+    """Send a command to the PTY and collect output until a marker appears."""
+
+    await session.stream.send(command.encode())
+    return await collect_output_until(session, marker, timeout=timeout)
+
+
 async def main() -> int:
     print("=" * 60)
     print("Low-level AsyncPTYSession Example")
@@ -83,11 +98,32 @@ async def main() -> int:
             print(initial_output_text.rstrip())
             print("-" * 60)
 
+            print("Verifying PTY input/output round-trip...")
+            try:
+                ready_output = await run_and_collect(
+                    session,
+                    f"printf '{READY_OUTPUT}\\n'\n",
+                    READY_OUTPUT,
+                )
+            except asyncio.TimeoutError:
+                print("Timed out waiting for PTY round-trip confirmation.")
+                return 1
+
+            ready_output_text = ready_output.decode("utf-8", errors="replace")
+            print()
+            print("Received PTY round-trip output:")
+            print("-" * 60)
+            print(ready_output_text.rstrip())
+            print("-" * 60)
+
             print("Writing a simple command through the PTY stream...")
-            await session.stream.send(f"printf '{EXPECTED_OUTPUT}\\n'; pwd; exit\n".encode())
 
             try:
-                output = await collect_output_until(session, EXPECTED_OUTPUT)
+                output = await run_and_collect(
+                    session,
+                    f"printf '{EXPECTED_OUTPUT}\\n'; pwd; printf '{DONE_OUTPUT}\\n'; exit\n",
+                    DONE_OUTPUT,
+                )
             except asyncio.TimeoutError:
                 print("Timed out waiting for PTY output.")
                 return 1
@@ -102,6 +138,9 @@ async def main() -> int:
             if EXPECTED_OUTPUT not in output_text:
                 print("Expected PTY output marker was not observed.")
                 return 1
+            if DONE_OUTPUT not in output_text:
+                print("PTY command completion marker was not observed.")
+                return 1
 
     print()
     print("Low-level PTY session flow completed successfully.")

From 4242bcbd154d9fee96d4c142ac35cf762604db48 Mon Sep 17 00:00:00 2001
From: Scott Trinh <scott@scotttrinh.com>
Date: Wed, 15 Apr 2026 16:55:22 -0400
Subject: [PATCH 2/2] Avoid echoed PTY sentinel matches

The readiness and completion sentinels in the PTY example could
match the shell's echoed command line before the command actually
executed.

Require sentinel markers to appear as complete output lines so the
example waits for real PTY output before continuing or exiting.
---
 examples/sandbox_14_pty_test.py | 17 +++++++++++++++--
 1 file changed, 15 insertions(+), 2 deletions(-)

diff --git a/examples/sandbox_14_pty_test.py b/examples/sandbox_14_pty_test.py
index 4a569549..aec9b798 100644
--- a/examples/sandbox_14_pty_test.py
+++ b/examples/sandbox_14_pty_test.py
@@ -30,14 +30,19 @@ async def collect_output_until(
     marker: str,
     *,
     timeout: float = 30.0,
+    require_full_line: bool = False,
 ) -> bytes:
     """Read PTY output until the expected marker appears."""
 
     async def _collect() -> bytes:
         output = b""
+        marker_bytes = marker.encode()
         async for data in session.stream:
             output += data
-            if marker.encode() in output:
+            if require_full_line:
+                if any(line.rstrip(b"\r") == marker_bytes for line in output.splitlines()):
+                    return output
+            elif marker_bytes in output:
                 return output
         return output
 
@@ -50,11 +55,17 @@ async def run_and_collect(
     marker: str,
     *,
     timeout: float = 30.0,
+    require_full_line: bool = False,
 ) -> bytes:
     """Send a command to the PTY and collect output until a marker appears."""
 
     await session.stream.send(command.encode())
-    return await collect_output_until(session, marker, timeout=timeout)
+    return await collect_output_until(
+        session,
+        marker,
+        timeout=timeout,
+        require_full_line=require_full_line,
+    )
 
 
 async def main() -> int:
@@ -104,6 +115,7 @@ async def main() -> int:
                     session,
                     f"printf '{READY_OUTPUT}\\n'\n",
                     READY_OUTPUT,
+                    require_full_line=True,
                 )
             except asyncio.TimeoutError:
                 print("Timed out waiting for PTY round-trip confirmation.")
@@ -123,6 +135,7 @@ async def main() -> int:
                     session,
                     f"printf '{EXPECTED_OUTPUT}\\n'; pwd; printf '{DONE_OUTPUT}\\n'; exit\n",
                     DONE_OUTPUT,
+                    require_full_line=True,
                 )
             except asyncio.TimeoutError:
                 print("Timed out waiting for PTY output.")