From f7c412ce39521c3171ef75c16640bd798deaa1b0 Mon Sep 17 00:00:00 2001 From: Albert Chen Date: Sun, 16 Aug 2026 15:06:11 +0800 Subject: [PATCH 1/2] fix: fix indefinite ParaTest hang: `stream_set_timeout()` does not apply to writes --- src/prompts/src/Support/Logger.php | 17 +++++- src/prompts/src/Support/Utils.php | 91 +++++++++++++++++++++++------- src/prompts/src/Task.php | 14 +++-- tests/Prompts/LoggerTest.php | 36 ++++++++++-- 4 files changed, 126 insertions(+), 32 deletions(-) diff --git a/src/prompts/src/Support/Logger.php b/src/prompts/src/Support/Logger.php index 082a00322..411b3ccc4 100644 --- a/src/prompts/src/Support/Logger.php +++ b/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; diff --git a/src/prompts/src/Support/Utils.php b/src/prompts/src/Support/Utils.php index f946ac4eb..f6d0c2a17 100644 --- a/src/prompts/src/Support/Utils.php +++ b/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,83 @@ 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; + } + + // Selectable streams are driven non-blocking; the rest keep their original mode. + $selectable = stream_get_meta_data($stream)['stream_type'] !== self::MEMORY_STREAM_TYPE + && @stream_set_blocking($stream, false); $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 { + if ($selectable) { + @stream_set_blocking($stream, 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), + ); + } } diff --git a/src/prompts/src/Task.php b/src/prompts/src/Task.php index d67d8738d..2405f0364 100644 --- a/src/prompts/src/Task.php +++ b/src/prompts/src/Task.php @@ -23,7 +23,7 @@ class Task extends Prompt * * This must remain well above the public render interval. */ - protected const 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; } diff --git a/tests/Prompts/LoggerTest.php b/tests/Prompts/LoggerTest.php index c656a53a8..56a5fe356 100644 --- a/tests/Prompts/LoggerTest.php +++ b/tests/Prompts/LoggerTest.php @@ -5,6 +5,7 @@ namespace Hypervel\Tests\Prompts; use Hypervel\Prompts\Support\Logger; +use Hypervel\Prompts\Support\Utils; use Hypervel\Tests\TestCase; use ReflectionProperty; @@ -84,8 +85,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 +113,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 +136,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 +149,35 @@ 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 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'); From 644e1cc3fbeb3d1254d61396555a1aee0ba94aec Mon Sep 17 00:00:00 2001 From: Albert Chen Date: Sun, 16 Aug 2026 15:22:05 +0800 Subject: [PATCH 2/2] fix: preserve the caller's non-blocking stream mode in writeAll() --- src/prompts/src/Support/Utils.php | 19 +++++++++++++++++-- tests/Prompts/LoggerTest.php | 25 +++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/src/prompts/src/Support/Utils.php b/src/prompts/src/Support/Utils.php index f6d0c2a17..c77f87924 100644 --- a/src/prompts/src/Support/Utils.php +++ b/src/prompts/src/Support/Utils.php @@ -95,9 +95,12 @@ public static function writeAll($stream, string $payload, float $timeout = 10.0) return; } + $metadata = stream_get_meta_data($stream); + // Selectable streams are driven non-blocking; the rest keep their original mode. - $selectable = stream_get_meta_data($stream)['stream_type'] !== self::MEMORY_STREAM_TYPE + $selectable = $metadata['stream_type'] !== self::MEMORY_STREAM_TYPE && @stream_set_blocking($stream, false); + $blocking = self::isBlocking($metadata); $offset = 0; try { @@ -130,12 +133,24 @@ public static function writeAll($stream, string $payload, float $timeout = 10.0) } } } finally { - if ($selectable) { + // 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 $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. * diff --git a/tests/Prompts/LoggerTest.php b/tests/Prompts/LoggerTest.php index 56a5fe356..e07a9d062 100644 --- a/tests/Prompts/LoggerTest.php +++ b/tests/Prompts/LoggerTest.php @@ -8,6 +8,7 @@ use Hypervel\Prompts\Support\Utils; use Hypervel\Tests\TestCase; use ReflectionProperty; +use RuntimeException; class LoggerTest extends TestCase { @@ -165,6 +166,30 @@ public function testRestoresTheOriginalBlockingModeAfterATimedOutWrite(): void } } + 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+');