coop/rose-ash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix pipe desync: async drain on _send, robust Playwright tests

Browse Source 
Root cause: OcamlBridge._send() used write() without drain().
asyncio.StreamWriter buffers writes — without drain(), multiple
commands accumulate and flush as a batch. The kernel processes
them sequentially, sending responses, but Python only reads one
response per command → pipe desync → "unexpected response" errors.

Fix: _send() is now async, calls drain() after every write.
All 14 callers updated to await.

Playwright tests rewritten:
- test_home_has_header: verifies #logo-opacity visible (was only
  checking for "sx" text — never caught missing header)
- test_home_has_nav_children: Geography link must be visible
- test_home_has_main_panel: #main-panel must have child elements
- TestDirectPageLoad: fresh browser.new_context() per test to
  avoid stale component hash in localStorage
- _setup_error_capture + _check_no_fatal_errors helpers

_render_to_sx uses aser_slot (not aser) — layout wrappers contain
re-parsed content that needs full expansion capability.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
giles
2026-03-19 16:11:55 +00:00
parent f819fda587
commit d3b3b4b720
4 changed files with 102 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
shared/sx/ocaml_bridge.py
View File

@@ -74,7 +74,7 @@ class OcamlBridge:
self._started = True
# Verify engine identity
self._send("(ping)")
await self._send("(ping)")
kind, engine = await self._read_response()
engine_name = engine if kind == "ok" else "unknown"
_logger.info("OCaml SX kernel ready (pid=%d, engine=%s)", self._proc.pid, engine_name)
@@ -95,21 +95,21 @@ class OcamlBridge:
async def ping(self) -> str:
"""Health check — returns engine name (e.g. 'ocaml-cek')."""
async with self._lock:
self._send("(ping)")
await self._send("(ping)")
kind, value = await self._read_response()
return value or "" if kind == "ok" else ""
async def load(self, path: str) -> int:
"""Load an .sx file for side effects (defcomp, define, defmacro)."""
async with self._lock:
self._send(f'(load "{_escape(path)}")')
await self._send(f'(load "{_escape(path)}")')
value = await self._read_until_ok(ctx=None)
return int(float(value)) if value else 0
async def load_source(self, source: str) -> int:
"""Evaluate SX source for side effects."""
async with self._lock:
self._send(f'(load-source "{_escape(source)}")')
await self._send(f'(load-source "{_escape(source)}")')
value = await self._read_until_ok(ctx=None)
return int(float(value)) if value else 0
@@ -121,7 +121,7 @@ class OcamlBridge:
"""
await self._ensure_components()
async with self._lock:
self._send(f'(eval "{_escape(source)}")')
await self._send(f'(eval "{_escape(source)}")')
return await self._read_until_ok(ctx)
async def render(
@@ -132,14 +132,14 @@ class OcamlBridge:
"""Render SX to HTML, handling io-requests via Python async IO."""
await self._ensure_components()
async with self._lock:
self._send(f'(render "{_escape(source)}")')
await self._send(f'(render "{_escape(source)}")')
return await self._read_until_ok(ctx)
async def aser(self, source: str, ctx: dict[str, Any] | None = None) -> str:
"""Evaluate SX and return SX wire format, handling io-requests."""
await self._ensure_components()
async with self._lock:
self._send(f'(aser "{_escape(source)}")')
await self._send(f'(aser "{_escape(source)}")')
return await self._read_until_ok(ctx)
async def aser_slot(self, source: str, ctx: dict[str, Any] | None = None) -> str:
@@ -154,7 +154,7 @@ class OcamlBridge:
# a separate lock acquisition could let another coroutine
# interleave commands between injection and aser-slot.
await self._inject_helpers_locked()
self._send(f'(aser-slot "{_escape(source)}")')
await self._send(f'(aser-slot "{_escape(source)}")')
return await self._read_until_ok(ctx)
async def _inject_helpers_locked(self) -> None:
@@ -183,7 +183,7 @@ class OcamlBridge:
arg_list = " ".join(chr(97 + i) for i in range(nargs))
sx_def = f'(define {name} (fn ({param_names}) (helper "{name}" {arg_list})))'
try:
self._send(f'(load-source "{_escape(sx_def)}")')
await self._send(f'(load-source "{_escape(sx_def)}")')
await self._read_until_ok(ctx=None)
count += 1
except OcamlBridgeError:
@@ -252,7 +252,7 @@ class OcamlBridge:
async with self._lock:
for filepath in all_files:
try:
self._send(f'(load "{_escape(filepath)}")')
await self._send(f'(load "{_escape(filepath)}")')
value = await self._read_until_ok(ctx=None)
# Response may be a number (count) or a value — just count files
count += 1
@@ -279,7 +279,7 @@ class OcamlBridge:
if callable(fn) and not name.startswith("~"):
sx_def = f'(define {name} (fn (&rest args) (apply helper (concat (list "{name}") args))))'
try:
self._send(f'(load-source "{_escape(sx_def)}")')
await self._send(f'(load-source "{_escape(sx_def)}")')
await self._read_until_ok(ctx=None)
count += 1
except OcamlBridgeError:
@@ -290,7 +290,7 @@ class OcamlBridge:
async def reset(self) -> None:
"""Reset the kernel environment to pristine state."""
async with self._lock:
self._send("(reset)")
await self._send("(reset)")
kind, value = await self._read_response()
if kind == "error":
raise OcamlBridgeError(f"reset: {value}")
@@ -299,10 +299,12 @@ class OcamlBridge:
# Internal protocol handling
# ------------------------------------------------------------------
def _send(self, line: str) -> None:
"""Write a line to the subprocess stdin."""
async def _send(self, line: str) -> None:
"""Write a line to the subprocess stdin and flush."""
assert self._proc and self._proc.stdin
_logger.debug("SEND: %s", line[:120])
self._proc.stdin.write((line + "\n").encode())
await self._proc.stdin.drain()
async def _readline(self) -> str:
"""Read a line from the subprocess stdout."""
@@ -316,7 +318,9 @@ class OcamlBridge:
raise OcamlBridgeError(
f"OCaml subprocess died unexpectedly. stderr: {stderr.decode(errors='replace')}"
)
return data.decode().rstrip("\n")
line = data.decode().rstrip("\n")
_logger.debug("RECV: %s", line[:120])
return line
async def _read_response(self) -> tuple[str, str | None]:
"""Read a single (ok ...) or (error ...) response.
@@ -341,11 +345,11 @@ class OcamlBridge:
if line.startswith("(io-request "):
try:
result = await self._handle_io_request(line, ctx)
self._send(f"(io-response {_serialize_for_ocaml(result)})")
await self._send(f"(io-response {_serialize_for_ocaml(result)})")
except Exception as e:
# MUST send a response or the pipe desyncs
_logger.warning("IO request failed, sending nil: %s", e)
self._send("(io-response nil)")
await self._send("(io-response nil)")
continue
kind, value = _parse_response(line)