fix: fix indefinite ParaTest hang: stream_set_timeout() does not apply to writes

src/prompts/src/Support/Logger.php

@@ -8,6 +8,13 @@
 
 class Logger
 {
+    /**
+     * How long a write may wait for the renderer to accept output.
+     *
+     * This must remain well above the public render interval.
+     */
+    public const float DEFAULT_WRITE_TIMEOUT_SECONDS = 10.0;
+
     /**
      * The first transport failure encountered while writing.
      */
@@ -17,9 +24,13 @@ class Logger
      * Create a new Logger instance.
      *
      * @param null|resource $socket
+     * @param float $writeTimeout seconds a write may wait for the renderer to accept output
      */
-    public function __construct(protected string $identifier, protected $socket = null)
-    {
+    public function __construct(
+        protected string $identifier,
+        protected $socket = null,
+        protected float $writeTimeout = self::DEFAULT_WRITE_TIMEOUT_SECONDS,
+    ) {
     }
 
     /**
@@ -110,7 +121,7 @@ protected function write(string $message, ?string $type = null): void
 
         try {
             // Each protocol frame must be complete because stream writes may be partial.
-            Utils::writeAll($this->socket, $payload);
+            Utils::writeAll($this->socket, $payload, $this->writeTimeout);
         } catch (RuntimeException $exception) {
             $this->transportFailure ??= $exception;
             $this->socket = null;

src/prompts/src/Support/Utils.php

@@ -12,6 +12,16 @@
  */
 class Utils
 {
+    /**
+     * The largest chunk offered to a single write attempt.
+     */
+    private const int WRITE_CHUNK_BYTES = 65536;
+
+    /**
+     * The stream type that always accepts a full write and cannot be selected on.
+     */
+    private const string MEMORY_STREAM_TYPE = 'MEMORY';
+
     /**
      * Determine if all items in an array match a truth test.
      *
@@ -67,40 +77,98 @@ public static function stripEscapeSequences(string $text): string
     }
 
     /**
-     * Write an entire payload to a stream.
+     * Write an entire payload to a stream, waiting up to a timeout for a blocked reader.
+     *
+     * The stream is driven in non-blocking mode because stream_set_timeout() only
+     * governs reads. A blocking write into a full buffer parks in the underlying
+     * send syscall on macOS and BSD, where no timeout applies and no signal can
+     * interrupt it, so the caller would wait for the reader forever.
      *
      * @param resource $stream
+     * @param float $timeout seconds to wait for a stalled reader to accept more output
      */
-    public static function writeAll($stream, string $payload): void
+    public static function writeAll($stream, string $payload, float $timeout = 10.0): void
     {
         $length = strlen($payload);
+
+        if ($length === 0) {
+            return;
+        }
+
+        $metadata = stream_get_meta_data($stream);
+
+        // Selectable streams are driven non-blocking; the rest keep their original mode.
+        $selectable = $metadata['stream_type'] !== self::MEMORY_STREAM_TYPE
+            && @stream_set_blocking($stream, false);
+        $blocking = self::isBlocking($metadata);
         $offset = 0;
 
-        while ($offset < $length) {
-            $written = @fwrite($stream, substr($payload, $offset));
+        try {
+            while ($offset < $length) {
+                // Capping the chunk keeps the copy proportional to what a write can accept.
+                $written = @fwrite($stream, substr($payload, $offset, self::WRITE_CHUNK_BYTES));
 
-            if (is_int($written) && $written > 0) {
-                $offset += $written;
+                if (is_int($written) && $written > 0) {
+                    $offset += $written;
 
-                if ($offset === $length) {
                     continue;
                 }
-            }
 
-            $metadata = stream_get_meta_data($stream);
+                // Only a failed write reports closure; zero simply means the buffer is full.
+                if ($written === false) {
+                    throw new RuntimeException(
+                        feof($stream)
+                            ? 'The prompt renderer closed while receiving output.'
+                            : 'Unable to write output to the prompt renderer.',
+                    );
+                }
 
-            // A timed-out stream stays latched and must not be reused after this failure.
-            if ($metadata['timed_out']) {
-                throw new RuntimeException('The prompt renderer timed out while receiving output.');
-            }
+                // A blocking stream returning zero cannot be waited on without stalling.
+                if (! $selectable) {
+                    throw new RuntimeException('Unable to write output to the prompt renderer.');
+                }
 
-            if ($written === false || $written === 0) {
-                throw new RuntimeException(
-                    $metadata['eof']
-                        ? 'The prompt renderer closed while receiving output.'
-                        : 'Unable to write output to the prompt renderer.',
-                );
+                if (! self::awaitWritable($stream, $timeout)) {
+                    throw new RuntimeException('The prompt renderer timed out while receiving output.');
+                }
+            }
+        } finally {
+            // Only restore blocking mode when the caller was relying on it.
+            if ($selectable && $blocking) {
+                @stream_set_blocking($stream, true);
             }
         }
     }
+
+    /**
+     * Determine whether a stream was in blocking mode.
+     *
+     * @param array<string, mixed> $metadata
+     */
+    private static function isBlocking(array $metadata): bool
+    {
+        // php://temp omits the blocking state entirely, so treat it as the stream default.
+        return (bool) ($metadata['blocked'] ?? true);
+    }
+
+    /**
+     * Wait for a stream to accept more output.
+     *
+     * @param resource $stream
+     */
+    private static function awaitWritable($stream, float $timeout): bool
+    {
+        $read = null;
+        $except = null;
+        $write = [$stream];
+        $seconds = (int) $timeout;
+
+        return (bool) @stream_select(
+            $read,
+            $write,
+            $except,
+            $seconds,
+            (int) round(($timeout - $seconds) * 1_000_000),
+        );
+    }
 }

src/prompts/src/Task.php

@@ -23,7 +23,7 @@ class Task extends Prompt
      *
      * This must remain well above the public render interval.
      */
-    protected const int LOGGER_WRITE_TIMEOUT_SECONDS = 10;
+    protected const float LOGGER_WRITE_TIMEOUT_SECONDS = Logger::DEFAULT_WRITE_TIMEOUT_SECONDS;
 
     /**
      * Scheduling margin added to one complete renderer frame interval.
@@ -517,10 +517,9 @@ protected function renderInProcess(Closure $callback): mixed
         fclose($sockets[0]);
         $this->pid = $pid;
         $this->socket = $sockets[1];
-        stream_set_timeout($this->socket, static::LOGGER_WRITE_TIMEOUT_SECONDS);
 
         $rendererInterval = $this->interval;
-        $logger = new Logger($this->identifier, $this->socket);
+        $logger = new Logger($this->identifier, $this->socket, static::LOGGER_WRITE_TIMEOUT_SECONDS);
         $result = null;
         $callbackFailure = null;
         $rendererFailure = null;
@@ -539,7 +538,11 @@ protected function renderInProcess(Closure $callback): mixed
         if ($rendererFailure === null) {
             try {
                 // A truncated newline-framed message makes a later reset unrecoverable.
-                Utils::writeAll($this->socket, $this->identifier . '_reset:' . ($success ? '1' : '0') . PHP_EOL);
+                Utils::writeAll(
+                    $this->socket,
+                    $this->identifier . '_reset:' . ($success ? '1' : '0') . PHP_EOL,
+                    static::LOGGER_WRITE_TIMEOUT_SECONDS,
+                );
                 $settlementTimeout = $rendererInterval + static::RENDERER_SETTLEMENT_MARGIN_MILLISECONDS;
                 stream_set_timeout(
                     $this->socket,
@@ -628,8 +631,7 @@ protected function runRendererProcess($socket): never
                 usleep($this->interval * 1000);
             }
 
-            stream_set_blocking($socket, true);
-            Utils::writeAll($socket, static::RENDERER_ACKNOWLEDGEMENT);
+            Utils::writeAll($socket, static::RENDERER_ACKNOWLEDGEMENT, static::LOGGER_WRITE_TIMEOUT_SECONDS);
         } catch (Throwable) {
             $exitCode = 1;
         }

tests/Prompts/LoggerTest.php

@@ -5,8 +5,10 @@
 namespace Hypervel\Tests\Prompts;
 
 use Hypervel\Prompts\Support\Logger;
+use Hypervel\Prompts\Support\Utils;
 use Hypervel\Tests\TestCase;
 use ReflectionProperty;
+use RuntimeException;
 
 class LoggerTest extends TestCase
 {
@@ -84,8 +86,7 @@ public function testPeerClosureLatchesFailureAndStopsLaterWrites(): void
     public function testNoReaderTimesOutAfterOneWindowFollowingAPartialWrite(): void
     {
         $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
-        stream_set_timeout($sockets[0], 1);
-        $logger = new Logger('abc123', $sockets[0]);
+        $logger = new Logger('abc123', $sockets[0], 1.0);
         $payload = str_repeat('x', 8 * 1024 * 1024);
         $startedAt = hrtime(true);
 
@@ -113,7 +114,6 @@ public function testNoReaderTimesOutAfterOneWindowFollowingAPartialWrite(): void
     public function testProgressingReaderMayExceedTheNoProgressWindow(): void
     {
         $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
-        stream_set_timeout($sockets[0], 1);
         $payload = str_repeat('x', 2 * 1024 * 1024);
         $pid = pcntl_fork();
 
@@ -137,7 +137,7 @@ public function testProgressingReaderMayExceedTheNoProgressWindow(): void
 
         $this->assertGreaterThan(0, $pid);
         fclose($sockets[1]);
-        $logger = new Logger('abc123', $sockets[0]);
+        $logger = new Logger('abc123', $sockets[0], 1.0);
         $startedAt = hrtime(true);
         $logger->line($payload);
         fclose($sockets[0]);
@@ -150,6 +150,59 @@ public function testProgressingReaderMayExceedTheNoProgressWindow(): void
         $this->assertSame(0, pcntl_wexitstatus($status));
     }
 
+    public function testRestoresTheOriginalBlockingModeAfterATimedOutWrite(): void
+    {
+        $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        $logger = new Logger('abc123', $sockets[0], 0.5);
+
+        try {
+            $logger->line(str_repeat('x', 1024 * 1024));
+
+            $this->assertNotNull($logger->transportFailure());
+            $this->assertTrue(stream_get_meta_data($sockets[0])['blocked']);
+        } finally {
+            fclose($sockets[0]);
+            fclose($sockets[1]);
+        }
+    }
+
+    public function testLeavesANonBlockingStreamNonBlocking(): void
+    {
+        $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+        stream_set_blocking($sockets[0], false);
+
+        try {
+            Utils::writeAll($sockets[0], "output\n", 0.5);
+
+            $this->assertFalse(stream_get_meta_data($sockets[0])['blocked']);
+
+            // A timed-out write must not silently hand back a blocking stream either.
+            try {
+                Utils::writeAll($sockets[0], str_repeat('x', 1024 * 1024), 0.5);
+            } catch (RuntimeException) {
+                // The timeout is the point of the payload; the mode is what matters here.
+            }
+
+            $this->assertFalse(stream_get_meta_data($sockets[0])['blocked']);
+        } finally {
+            fclose($sockets[0]);
+            fclose($sockets[1]);
+        }
+    }
+
+    public function testWritesAnEntirePayloadToAnInMemoryStream(): void
+    {
+        $stream = fopen('php://memory', 'w+');
+        $payload = str_repeat('prompt output ', 1024);
+
+        Utils::writeAll($stream, $payload);
+        rewind($stream);
+
+        $this->assertSame($payload, stream_get_contents($stream));
+
+        fclose($stream);
+    }
+
     public function testDoesNotThrowWhenConstructedWithoutSocket(): void
     {
         $logger = new Logger('abc123');