feat(native): add JSONL streaming reader for native units (W7 contract gap)

stream_native(): per-line json.loads over subprocess stdout, progress
events yielded live, terminal envelope validated as the FINAL line,
watchdog timeout with adapter-kill vs unit-death distinction, concurrent
stderr drain. Generator yields event dicts then one terminal NativeResult
(gaining an events field). Malformed interior lines skipped and counted;
a malformed final line invalidates the envelope, so ok=False while partial
progress stays available. Tests use a Python subprocess as the fake
native unit — no Rust build dependency in the test path (house style
from tests/test_native_interop_proof.py).

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,
+    )
+

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"]))