diff --git a/shared/native_units.py b/shared/native_units.py index e3bd5b89..0f54901e 100644 --- a/shared/native_units.py +++ b/shared/native_units.py @@ -16,6 +16,7 @@ from __future__ import annotations import json import subprocess +import threading import time from dataclasses import dataclass, field from pathlib import Path @@ -36,6 +37,8 @@ class NativeResult: stderr — captured stderr text (may be non-empty even on success) duration_ms — wall-clock cost of the invocation timed_out — True when the unit exceeded the timeout and was killed + events — in streaming mode: every parsed JSON object line, in + arrival order (empty for non-streaming :func:`run_native`) """ ok: bool @@ -45,6 +48,7 @@ class NativeResult: duration_ms: int timed_out: bool = False raw_stdout: str = field(default="", repr=False) + events: list[dict] = field(default_factory=list) def run_native(binary: Path, args: list[str], timeout_s: float = 10.0) -> NativeResult: @@ -101,3 +105,112 @@ def run_native(binary: Path, args: list[str], timeout_s: float = 10.0) -> Native def run_proof(args: list[str], timeout_s: float = 10.0) -> NativeResult: """Run the Wave 0 proof unit with the standard contract defaults.""" return run_native(PROOF_UNIT, args, timeout_s=timeout_s) + + +def stream_native( + binary: Path, + args: list[str], + timeout_s: float = 10.0, + on_event=None, +): + """Run a native unit in JSONL streaming mode, yielding events as they arrive. + + The JSONL streaming contract (rubric §W7): one JSON object per stdout + line — ``{"ok": true, "event": ...}`` progress objects along the way, and + a terminal envelope as the FINAL line. This generator yields each parsed + event dict the moment it arrives (so a long op surfaces partial progress + instead of blocking until exit), then yields exactly one terminal + :class:`NativeResult`. *on_event*, when given, is called with each parsed + event as it arrives (the TUI wraps this in ``call_from_thread``). + + Stream-mode ok-semantics are stricter than :func:`run_native`: ``ok`` + requires exit 0, no timeout, AND a valid terminal envelope (the final + stdout line parsed as a JSON object whose ``ok`` is not False). Malformed + interior lines are skipped and counted in ``malformed_lines``; malformed + FINAL line means no valid terminal envelope, so ``ok`` is False — partial + progress stays available in ``events``. That is the point of streaming. + + Same failure discipline as :func:`run_native`: raises FileNotFoundError + when *binary* is absent; every unit-level failure (bad envelope, timeout, + non-zero exit) is data on the terminal result, never an exception. + """ + if on_event is not None and not callable(on_event): + raise TypeError("on_event must be callable") + if not binary.is_file(): + raise FileNotFoundError(f"native unit not built: {binary}") + + started = time.monotonic() + proc = subprocess.Popen( + [str(binary), *args], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + # stderr must drain concurrently — reading it after the stdout loop + # deadlocks when a unit floods stderr past the pipe buffer mid-stream. + stderr_parts: list[str] = [] + stderr_drain = threading.Thread( + target=lambda: stderr_parts.append(proc.stderr.read()), daemon=True + ) + stderr_drain.start() + + killed = threading.Event() + # The watchdog's kill and the flag are one step so an adapter-side + # timeout is never mistaken for a unit dying by its own signal. + watchdog = threading.Timer(timeout_s, lambda: (killed.set(), proc.kill())) + watchdog.start() + + events: list[dict] = [] + malformed_lines = 0 + # True only while the most recent non-empty stdout line parsed as a JSON + # object — a malformed final line invalidates the terminal envelope. + final_line_is_json = False + try: + for line in proc.stdout: + line = line.strip() + if not line: + continue + try: + parsed = json.loads(line) + except json.JSONDecodeError: + malformed_lines += 1 + final_line_is_json = False + continue + if not isinstance(parsed, dict): + malformed_lines += 1 + final_line_is_json = False + continue + events.append(parsed) + final_line_is_json = True + if on_event is not None: + on_event(parsed) + yield parsed + proc.wait() + finally: + watchdog.cancel() + stderr_drain.join(timeout=5.0) + + timed_out = killed.is_set() + duration_ms = int((time.monotonic() - started) * 1000) + exit_code = proc.returncode if proc.returncode is not None else -1 + if timed_out: + exit_code = -1 # adapter-side kill, never a unit's own exit + + terminal = events[-1] if (events and final_line_is_json) else None + ok = ( + not timed_out + and exit_code == 0 + and terminal is not None + and terminal.get("ok") is not False + ) + yield NativeResult( + ok=ok, + exit_code=exit_code, + data=terminal, + stderr=stderr_parts[0] if stderr_parts else "", + duration_ms=duration_ms, + timed_out=timed_out, + events=events, + ) + diff --git a/tests/test_native_streaming.py b/tests/test_native_streaming.py new file mode 100644 index 00000000..5b58eb3d --- /dev/null +++ b/tests/test_native_streaming.py @@ -0,0 +1,131 @@ +"""JSONL streaming reader for native units (W7 contract gap, rubric §W7 point 5). + +Contract under test: a long-running native unit streams one JSON object per +line — ``{"ok":true,"event":...}`` progress objects — and ends with a terminal +envelope as the final stdout line. The adapter yields per-line events and +validates the terminal envelope. Fixtures are Python scripts standing in for +compiled units; the adapter treats any executable path identically. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT / "src")) +sys.path.insert(0, str(ROOT)) + +import pytest # noqa: E402 + +from shared.native_units import NativeResult, stream_native # noqa: E402 + +FIXTURE = """\ +#!/usr/bin/env python3 +import sys, time + +mode = sys.argv[1] if len(sys.argv) > 1 else "stream" + +if mode == "stream": + for i in range(3): + print(f'{{"ok":true,"event":"step-{i}"}}', flush=True) + time.sleep(0.02) + print('{"ok":true,"result":"done"}', flush=True) +elif mode == "slow": + print('{"ok":true,"event":"started"}', flush=True) + time.sleep(30) + print('{"ok":true,"result":"done"}', flush=True) +elif mode == "bad": + print('{"ok":true,"event":"one"}', flush=True) + print('not json at all', flush=True) + print('{"ok":true,"result":"done"}', flush=True) +elif mode == "junk-last": + print('{"ok":true,"event":"one"}', flush=True) + print('trailing noise', flush=True) +elif mode == "fail": + print('{"ok":false,"error":"boom"}', flush=True) + sys.exit(2) +""" + + +@pytest.fixture +def unit(tmp_path: Path) -> Path: + script = tmp_path / "stream-unit.py" + script.write_text(FIXTURE) + script.chmod(0o755) + return script + + +def _split(yielded: list) -> tuple[list[dict], NativeResult]: + terminal = yielded[-1] + assert isinstance(terminal, NativeResult), f"last yield must be the terminal NativeResult, got {type(terminal)}" + events = [item for item in yielded if isinstance(item, dict)] + return events, terminal + + +def test_streaming_yields_events_then_terminal_result(unit: Path): + yielded = list(stream_native(unit, ["stream"])) + events, terminal = _split(yielded) + # 3 progress events + the terminal envelope (also a parsed dict line) + assert [e.get("event") for e in events] == ["step-0", "step-1", "step-2", None] + assert events[-1] == {"ok": True, "result": "done"} + assert terminal.ok + assert terminal.exit_code == 0 + assert terminal.data == {"ok": True, "result": "done"} + + +def test_events_arrive_before_process_exit(unit: Path): + """Progress is visible DURING the run, not after it — the point of streaming.""" + seen: list[str] = [] + for item in stream_native(unit, ["stream"]): + if isinstance(item, dict) and "event" in item: + seen.append(item["event"]) + assert seen == ["step-0", "step-1", "step-2"] + + +def test_timeout_kills_midstream(unit: Path): + yielded = list(stream_native(unit, ["slow"], timeout_s=0.5)) + events, terminal = _split(yielded) + assert terminal.timed_out + assert not terminal.ok + assert terminal.exit_code == -1 + # the first progress event was still delivered before the kill + assert events and events[0] == {"ok": True, "event": "started"} + + +def test_malformed_interior_lines_skipped(unit: Path): + yielded = list(stream_native(unit, ["bad"])) + events, terminal = _split(yielded) + # the malformed line is skipped; progress + terminal envelope remain + progress = [e for e in events if "event" in e] + assert [e["event"] for e in progress] == ["one"] + assert terminal.ok + assert terminal.data == {"ok": True, "result": "done"} + + +def test_trailing_noise_invalidates_terminal_envelope(unit: Path): + """JSON was emitted, but the FINAL line is not an envelope — contract violation.""" + yielded = list(stream_native(unit, ["junk-last"])) + events, terminal = _split(yielded) + assert not terminal.ok + assert events and events[0] == {"ok": True, "event": "one"} + + +def test_failure_envelope_is_data_not_exception(unit: Path): + yielded = list(stream_native(unit, ["fail"])) + events, terminal = _split(yielded) + assert not terminal.ok + assert terminal.exit_code == 2 + assert terminal.data == {"ok": False, "error": "boom"} + + +def test_events_attached_to_terminal_result(unit: Path): + yielded = list(stream_native(unit, ["stream"])) + _, terminal = _split(yielded) + assert len(terminal.events) == 4 # 3 progress + 1 terminal envelope + assert terminal.events[-1] == {"ok": True, "result": "done"} + + +def test_missing_binary_raises_adapter_misuse(tmp_path: Path): + with pytest.raises(FileNotFoundError): + list(stream_native(tmp_path / "no-such-binary", ["stream"]))