cli: batch/streaming mode (harel run) per SPEC §13.7

pyproject.toml

@@ -37,6 +37,7 @@ packages = ["src/harel"]
 [tool.ruff]
 line-length = 100
 target-version = "py311"
+extend-exclude = ["vendor"]  # the harel spec repo is a vendored submodule, not ours to lint
 
 [tool.ruff.lint]
 select = ["E", "F", "I", "UP", "B", "W", "C4"]

src/harel/__main__.py

@@ -0,0 +1,8 @@
+"""Entry point for ``python -m harel`` (mirrors the ``harel`` console script)."""
+
+from __future__ import annotations
+
+from .cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/harel/cli.py

@@ -9,9 +9,11 @@
 from __future__ import annotations
 
 import argparse
+import io
 import json
 import os
 import sys
+from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
 from typing import Any, cast
 
@@ -103,6 +105,10 @@ def add(cmd: str, **kw: Any) -> argparse.ArgumentParser:
     r = add("restore")
     r.add_argument("snapshot")
     r.set_defaults(cmd=cmd_restore)
+
+    run = add("run")
+    run.add_argument("source", nargs="?", default="-", help="'-' for stdin, or an NDJSON file")
+    run.set_defaults(cmd=cmd_run)
     return p
 
 
@@ -299,6 +305,76 @@ def cmd_export(args: argparse.Namespace, store: Store) -> int:
     return EXIT_OK
 
 
+# --- batch / streaming mode (SPEC §13.7) ------------------------------------
+def cmd_run(args: argparse.Namespace, store: Store) -> int:
+    """Drive many commands from NDJSON stdin against one store + virtual clock.
+
+    Each input line is a JSON array of argv tokens (one §13.3 command). For each
+    line, exactly one NDJSON result object is written to stdout in input order:
+    ``{ "ok": bool, "exit": int, "result": <value>, "error": {"message": str}? }``.
+    A failing line does not abort the stream; the process exit code is the first
+    non-zero line exit, else 0.
+    """
+    if args.source in (None, "-"):
+        lines = sys.stdin.read().splitlines()
+    else:
+        lines = Path(args.source).read_text(encoding="utf-8").splitlines()
+    parser = _build_parser()
+    first_nonzero = 0
+    for raw in lines:
+        line = raw.strip()
+        if not line:
+            continue
+        exit_code, result, error = _run_one(parser, store, line)
+        record: dict[str, Any] = {"ok": exit_code == 0, "exit": exit_code, "result": result}
+        if error is not None:
+            record["error"] = {"message": error}
+        print(json.dumps(record), flush=True)
+        if exit_code != 0 and first_nonzero == 0:
+            first_nonzero = exit_code
+    return first_nonzero
+
+
+def _run_one(
+    parser: argparse.ArgumentParser, store: Store, line: str
+) -> tuple[int, Any, str | None]:
+    """Execute one batch line; return (exit_code, result_value, error_message)."""
+    try:
+        argv = json.loads(line)
+        if not isinstance(argv, list) or not all(isinstance(t, str) for t in argv):
+            raise ValueError("each line must be a JSON array of strings")
+    except (ValueError, json.JSONDecodeError) as exc:
+        return EXIT_USAGE, None, str(exc)
+
+    out, err = io.StringIO(), io.StringIO()
+    try:
+        with redirect_stdout(out), redirect_stderr(err):
+            sub = parser.parse_args([*argv, "--json"])
+            if getattr(sub, "command", None) == "run":
+                return EXIT_USAGE, None, "nested 'run' is not allowed in batch mode"
+            rc = int(sub.cmd(sub, store))
+    except SystemExit as exc:  # argparse usage error
+        code = exc.code if isinstance(exc.code, int) else EXIT_USAGE
+        return code, None, (err.getvalue().strip() or "usage error")
+    except HarelError as exc:
+        return EXIT_OTHER, None, str(exc)
+
+    result = _parse_captured(out.getvalue())
+    message = (err.getvalue().strip() or None) if rc != 0 else None
+    return rc, result, message
+
+
+def _parse_captured(text: str) -> Any:
+    """A command's captured stdout as JSON when it is JSON, else the raw string/None."""
+    s = text.strip()
+    if not s:
+        return None
+    try:
+        return json.loads(s)
+    except json.JSONDecodeError:
+        return s
+
+
 # --- output helpers ---------------------------------------------------------
 def _state_json(host: Host, inst: Instance) -> dict[str, Any]:
     return {

tests/harness.py

@@ -8,9 +8,11 @@
 
 from __future__ import annotations
 
-import json
+import importlib.util
+import sys
 from dataclasses import dataclass
 from pathlib import Path
+from types import ModuleType
 from typing import Any
 
 from harel import Host
@@ -99,58 +101,34 @@ def cli_cases() -> list[Path]:
     return sorted(p for p in cli_dir.iterdir() if p.is_dir())
 
 
-# --- CLI case runner (SPEC §13.6) ------------------------------------------
+# --- CLI case runner (SPEC §13.6): true black box via the spec repo's runner --
 def run_cli_case(case_dir: Path) -> None:
-    """Run a CLI case's steps against a fresh temp store; assert exit + output."""
-    import io
-    import tempfile
-    from contextlib import redirect_stdout
+    """Run a CLI case **black-box** via the spec repo's reference runner (§13.6).
 
-    from harel import cli
-
-    spec = _load_yaml(case_dir / "cli.yaml")
-    tmpstore = tempfile.mkdtemp(prefix="harel-cli-")
-    for i, step in enumerate(spec.get("steps", [])):
-        argv = [_resolve_arg(case_dir, a) for a in step["run"]]
-        buf = io.StringIO()
-        with redirect_stdout(buf):
-            rc = cli.main(["--store", tmpstore, *argv])
-        _check_cli_expect(case_dir.name, i, rc, buf.getvalue(), step.get("expect") or {})
-
-
-def _resolve_arg(case_dir: Path, arg: str) -> str:
-    # A bare filename that exists in the case dir (e.g. machine.yaml) is resolved.
-    if "/" not in arg and (case_dir / arg).exists():
-        return str(case_dir / arg)
-    return arg
+    Invokes this package as a subprocess (``python -m harel``), so packaging and
+    entry-point regressions are caught — not an in-process import. Delegating to
+    ``vendor/harel/conformance/run_cli.py`` also avoids harness drift.
+    """
+    runner = _load_cli_runner()
+    rc = runner.main(
+        [
+            "--cmd",
+            f"{sys.executable} -m harel",
+            "--conformance-dir",
+            str(CONFORMANCE_DIR / "cli"),
+            case_dir.name,
+        ]
+    )
+    assert rc == 0, f"cli/{case_dir.name}: black-box CLI runner reported failure"
 
 
-def _check_cli_expect(
-    name: str, i: int, rc: int, stdout: str, expect: dict[str, Any]
-) -> None:
-    label = f"cli/{name} step {i}"
-    if "exit" in expect:
-        assert rc == expect["exit"], f"{label}: exit {rc} != {expect['exit']}"
-    if "json" in expect:
-        actual = json.loads(stdout) if stdout.strip() else None
-        _assert_subset(actual, expect["json"], label)
-    elif "stdout" in expect:
-        assert stdout == expect["stdout"], f"{label}: stdout mismatch"
-
-
-def _assert_subset(actual: Any, expected: Any, label: str) -> None:
-    if isinstance(expected, dict):
-        assert isinstance(actual, dict), f"{label}: expected object, got {type(actual)}"
-        for k, v in expected.items():
-            assert k in actual, f"{label}: missing key '{k}'"
-            _assert_subset(actual[k], v, label)
-    elif isinstance(expected, list):
-        assert isinstance(actual, list), f"{label}: expected list"
-        assert len(actual) == len(expected), f"{label}: list length"
-        for a, e in zip(actual, expected, strict=False):
-            _assert_subset(a, e, label)
-    else:
-        assert actual == expected, f"{label}: {actual!r} != {expected!r}"
+def _load_cli_runner() -> ModuleType:
+    path = SUITE_DIR / "conformance" / "run_cli.py"
+    spec = importlib.util.spec_from_file_location("harel_cli_runner", path)
+    assert spec is not None and spec.loader is not None, f"runner not found: {path}"
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
 
 
 # --- engine case runner -----------------------------------------------------

tests/test_cli_stream.py

@@ -0,0 +1,107 @@
+"""Unit tests for the batch/streaming CLI mode (SPEC §13.7)."""
+
+from __future__ import annotations
+
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+import harel.cli as cli
+
+TURNSTILE = """\
+id: turnstile
+events:
+  coin: { payload: { amount: { type: int, required: true } } }
+  push: {}
+top:
+  esvs:
+    fare: { type: int, init: 50 }
+  initial: { transition_to: locked }
+  states:
+    locked:
+      on_events:
+        coin: { transition_to: unlocked, guard: "event.payload.amount >= fare" }
+    unlocked:
+      on_events:
+        push: { transition_to: locked }
+"""
+
+
+def _run_batch(
+    tmp_path: Path,
+    lines: list[list[str]],
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> tuple[int, list[dict]]:
+    stdin = "".join(json.dumps(line) + "\n" for line in lines)
+    monkeypatch.setattr("sys.stdin", io.StringIO(stdin))
+    rc = cli.main(["--store", str(tmp_path / "store"), "run", "-"])
+    out = capsys.readouterr().out
+    results = [json.loads(x) for x in out.splitlines() if x.strip()]
+    return rc, results
+
+
+def test_stream_happy_path(tmp_path, monkeypatch, capsys):
+    machine = tmp_path / "m.yaml"
+    machine.write_text(TURNSTILE)
+    rc, results = _run_batch(
+        tmp_path,
+        [
+            ["new", "t1", str(machine)],
+            ["send", "t1", "coin", "--payload", "amount=100"],
+            ["state", "t1"],
+        ],
+        monkeypatch,
+        capsys,
+    )
+    assert rc == 0
+    assert [r["ok"] for r in results] == [True, True, True]
+    assert results[0]["result"]["config"] == ["locked"]
+    assert results[1]["result"]["config"] == ["unlocked"]
+    assert results[1]["result"]["published"] == []
+    assert results[2]["result"]["config"] == ["unlocked"]
+
+
+def test_stream_failure_does_not_abort(tmp_path, monkeypatch, capsys):
+    machine = tmp_path / "m.yaml"
+    machine.write_text(TURNSTILE)
+    rc, results = _run_batch(
+        tmp_path,
+        [
+            ["new", "t1", str(machine)],
+            ["send", "missing", "push"],   # not found -> exit 4
+            ["state", "t1"],               # still runs
+        ],
+        monkeypatch,
+        capsys,
+    )
+    # process exit is the first non-zero line exit (§13.7).
+    assert rc == 4
+    assert results[1] == {
+        "ok": False,
+        "exit": 4,
+        "result": None,
+        "error": {"message": results[1]["error"]["message"]},
+    }
+    assert results[1]["error"]["message"]  # a diagnostic was captured
+    assert results[2]["ok"] is True
+    assert results[2]["result"]["config"] == ["locked"]
+
+
+def test_stream_malformed_line(tmp_path, monkeypatch, capsys):
+    monkeypatch.setattr("sys.stdin", io.StringIO("not json\n"))
+    rc = cli.main(["--store", str(tmp_path / "store"), "run", "-"])
+    out = capsys.readouterr().out
+    rec = json.loads(out.strip())
+    assert rc == 2
+    assert rec["ok"] is False and rec["exit"] == 2 and rec["result"] is None
+
+
+def test_stream_rejects_nested_run(tmp_path, monkeypatch, capsys):
+    monkeypatch.setattr("sys.stdin", io.StringIO(json.dumps(["run", "-"]) + "\n"))
+    rc = cli.main(["--store", str(tmp_path / "store"), "run", "-"])
+    rec = json.loads(capsys.readouterr().out.strip())
+    assert rc == 2
+    assert rec["ok"] is False and rec["exit"] == 2

tests/test_conformance.py

@@ -59,7 +59,7 @@ def test_bundled_schema_matches_submodule() -> None:
 
 def test_suite_present() -> None:
     assert len(engine_cases()) == 22, "expected 22 engine cases"
-    assert len(cli_cases()) == 1, "expected 1 CLI case"
+    assert len(cli_cases()) == 2, "expected 2 CLI cases"
 
 
 @pytest.mark.parametrize("case", engine_cases(), ids=lambda c: c.name)