fix: MCP 主循环 stdin 批量行饥饿

select + buffered readline 会丢批量到达的行:一行 readline
可能把多行一起读进用户态 buffer,之后 select 再也看不到
空 fd,buffer 里的行永远不被读。客户端在 initialize 后
习惯把 notification + 下一请求连写,MCP 于是只答 initialize
就卡死(生产 2ff8975 活体验证两跑复现,faulthandler 证实
主线程停在 select)。改用非阻塞 fd + 自有 buffer 的行泵,
每行按到达顺序必达;附带回归测试(批量行全达 + EOF 残行)。
This commit is contained in:
host committed 2026-10-30 12:47:23 +08:00
1 parent 2ff8975f43
commit f85de63120
2 files changed
+142 -34

No files matched your search

+58
View File
@@ -359,5 +359,63 @@ class MCPTest(unittest.TestCase):
self.assertIn(name, names)
class LineReaderTest(unittest.TestCase):
"""Pipelined stdin lines must all arrive (no select starvation).
Regression test for the live failure where the MCP answered
initialize and then hung forever: the client wrote
notifications/initialized + tools/call back-to-back, one
readline() buffered both, and select() never saw the second
line again because the fd was drained.
"""
def test_pipelined_lines_all_delivered(self):
import time
from playwright_mcp import _LineReader
rfd, wfd = os.pipe()
try:
reader = _LineReader(rfd)
os.write(wfd, b"init\n")
lines, drained = reader.poll()
self.assertEqual(lines, ["init"])
self.assertFalse(drained)
# Second batch arrives as one write, like a real client
# sending notification + next request back-to-back.
os.write(wfd, b"notif\nnavigate\n")
# Simulate the drain-loop delay: both lines sit in the
# pipe before the next poll.
time.sleep(0.2)
got = []
deadline = time.time() + 2
while len(got) < 2 and time.time() < deadline:
lines, drained = reader.poll()
got.extend(lines)
if not lines:
time.sleep(0.02)
self.assertEqual(got, ["notif", "navigate"])
finally:
os.close(wfd)
os.close(rfd)
def test_partial_line_flushed_at_eof(self):
from playwright_mcp import _LineReader
rfd, wfd = os.pipe()
try:
reader = _LineReader(rfd)
os.write(wfd, b"partial-no-newline")
os.close(wfd)
lines, drained = reader.poll()
self.assertEqual(lines, [])
self.assertFalse(drained)
lines, drained = reader.poll()
self.assertEqual(lines, ["partial-no-newline"])
self.assertTrue(drained)
finally:
try:
os.close(rfd)
except OSError:
pass
if __name__ == "__main__":
unittest.main()