fix(cli): help/version/usage, distinct errors, explicit exit codes, ledger tolerance (M4)

loop/__main__.py

@@ -1,11 +1,67 @@
 from __future__ import annotations
 
 import json
+import re
 import sys
 from pathlib import Path
 
 from .contract import doctor_report
 
+_PROG = "python3 -m loop"
+
+_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect")
+
+# Read commands operate on an EXISTING contract dir; scaffold CREATES one, so it
+# is exempt from the "target must exist" guard.
+_READ_COMMANDS = ("doctor", "validate", "verify", "inspect")
+
+_USAGE = f"usage: {_PROG} <scaffold|doctor|validate|verify|inspect> <workspace-or-.loop>"
+
+_HELP = f"""{_PROG} — validate and inspect a portable repo-OS loop contract.
+
+{_USAGE}
+
+commands:
+  scaffold   Write a fresh, doctor-clean loop contract into <target>.
+  doctor     Validate the contract objects (manifest, state, tasks, terminal).
+  validate   Alias for doctor.
+  verify     Alias for doctor — check the contract's state.
+  inspect    Score an existing loop against the prime-directive checklist
+             (emits a weak/strong verdict and a gap report).
+
+arguments:
+  <target>   A workspace root or its .loop/ directory.
+
+options:
+  -h, --help    Show this help and exit.
+  --version     Show the version and exit.
+"""
+
+
+def _version() -> str:
+    """Return the package version. Single source of truth is pyproject.toml.
+
+    Prefer installed metadata (which is itself generated from pyproject); fall
+    back to reading pyproject.toml at the repo root so `--version` still works
+    from an uninstalled/editable checkout.
+    """
+    try:
+        from importlib.metadata import PackageNotFoundError, version
+
+        try:
+            return version("loop-engineer")
+        except PackageNotFoundError:
+            pass
+    except Exception:  # pragma: no cover - importlib.metadata ships on 3.10+
+        pass
+    pyproject = Path(__file__).resolve().parent.parent / "pyproject.toml"
+    try:
+        text = pyproject.read_text(encoding="utf-8")
+    except OSError:  # pragma: no cover - repo layout guarantees this file
+        return "unknown"
+    match = re.search(r'(?m)^version\s*=\s*"([^"]+)"', text)
+    return match.group(1) if match else "unknown"
+
 
 def _print_json(report: dict) -> int:
     print(json.dumps(report, indent=2))
@@ -14,16 +70,38 @@ def _print_json(report: dict) -> int:
 
 def main(argv: list[str] | None = None) -> int:
     argv = list(sys.argv[1:] if argv is None else argv)
-    if not argv or argv[0] in {"-h", "--help"}:
+
+    if not argv:
+        print(_USAGE, file=sys.stderr)
+        return 2
+    if argv[0] in {"-h", "--help"}:
+        print(_HELP)
+        return 0
+    if argv[0] == "--version":
+        print(_version())
+        return 0
+
+    command = argv.pop(0)
+    if command not in _COMMANDS:
+        print(f"unknown loop command: {command}", file=sys.stderr)
+        print(_USAGE, file=sys.stderr)
+        return 2
+
+    if not argv:
+        print(f"{command}: missing target argument", file=sys.stderr)
+        print(_USAGE, file=sys.stderr)
+        return 2
+    target = Path(argv[0])
+
+    if command in _READ_COMMANDS and not target.exists():
         print(
-            "usage: python -m loop <scaffold|doctor|validate|verify|inspect> <workspace-or-.loop>",
+            f"{command}: target path does not exist: {target}\n"
+            f"       pass an existing workspace root or its .loop/ directory "
+            f"(run `{_PROG} scaffold {target}` to create a new contract).",
             file=sys.stderr,
         )
         return 2
 
-    command = argv.pop(0)
-    target = Path(argv[0]) if argv else Path.cwd()
-
     if command == "scaffold":
         from .scaffold import scaffold
 
@@ -38,19 +116,16 @@ def main(argv: list[str] | None = None) -> int:
     if command in {"doctor", "validate", "verify"}:
         return _print_json(doctor_report(target))
 
-    if command == "inspect":
-        # Keep the historical inspector script as the scoring UI over the same
-        # contract artifacts; import lazily to avoid making scripts/ a package.
-        scripts_dir = Path(__file__).resolve().parent.parent / "scripts"
-        sys.path.insert(0, str(scripts_dir))
-        import inspect_loop  # type: ignore
+    # command == "inspect": keep the historical inspector script as the scoring
+    # UI over the same contract artifacts; import lazily to avoid making
+    # scripts/ a package.
+    scripts_dir = Path(__file__).resolve().parent.parent / "scripts"
+    sys.path.insert(0, str(scripts_dir))
+    import inspect_loop  # type: ignore
 
-        report = inspect_loop.inspect_loop(str(target))
-        print(json.dumps(report, indent=2))
-        return 0 if report.get("verdict") != "weak" else 1
-
-    print(f"unknown loop command: {command}", file=sys.stderr)
-    return 2
+    report = inspect_loop.inspect_loop(str(target))
+    print(json.dumps(report, indent=2))
+    return 0 if report.get("verdict") != "weak" else 1
 
 
 if __name__ == "__main__":

scripts/anticheat_scan.py

@@ -24,8 +24,16 @@
     python3 anticheat_scan.py --diff changes.diff --trajectory trace.json
 
 ``--trajectory`` is a JSON array of strings (tool calls / paths the loop touched).
-Prints findings as JSON. Exit 0 = clean, 1 = downgrade (FailedUnverifiable),
-2 = gate tampering (FailedSafety).
+Prints findings as JSON. The exit code splits the scan's *outcome* from an
+*operational* failure so a broken invocation is never read as a clean scan or as
+findings:
+
+  * 0 — clean: the scan ran and found nothing.
+  * 1 — findings: a downgrade to FailedUnverifiable, or review-only findings.
+  * 2 — gate tampering: a critical finding -> FailedSafety.
+  * 3 — operational error: the scan could not run at all (unreadable ``--diff`` /
+        ``--trajectory`` file, malformed ``--trajectory`` JSON, failed
+        ``--self-check`` git call, or a bad argument).
 """
 
 from __future__ import annotations
@@ -34,8 +42,17 @@
 import json
 import pathlib
 import re
+import subprocess
 import sys
 
+# Exit codes. Findings (1, 2) and the scan's severity live apart from an
+# operational failure (3), so a broken invocation never masquerades as a clean
+# scan (0) or as a downgrade finding (1).
+EXIT_CLEAN = 0
+EXIT_FINDINGS = 1
+EXIT_GATE_TAMPERING = 2
+EXIT_OPERATIONAL_ERROR = 3
+
 # Path markers for a gate file. A bare name (e.g. "self_eval.py") matches by
 # BASENAME / path-segment, never by raw substring — so a test file whose name
 # merely CONTAINS a gate name (test_anticheat_scan.py) is not mis-upgraded.
@@ -411,8 +428,19 @@ def scan(
     return {"findings": findings, "clean": not findings, "downgrade_to": downgrade}
 
 
+class _ArgParser(argparse.ArgumentParser):
+    """Argparse variant that exits with the operational-error code on a bad
+    argument, instead of argparse's default 2 (which would collide with the
+    gate-tampering finding code)."""
+
+    def error(self, message: str):  # noqa: D401 - argparse override
+        self.print_usage(sys.stderr)
+        print(f"{self.prog}: error: {message}", file=sys.stderr)
+        raise SystemExit(EXIT_OPERATIONAL_ERROR)
+
+
 def main(argv: list[str]) -> int:
-    ap = argparse.ArgumentParser(description="Anti-cheat trajectory scan.")
+    ap = _ArgParser(description="Anti-cheat trajectory scan.")
     ap.add_argument("--diff", help="path to a unified diff file (else read stdin)")
     ap.add_argument("--files", help="comma-separated changed files (overrides diff parse)")
     ap.add_argument("--trajectory", help="path to a JSON array of trajectory strings")
@@ -423,34 +451,39 @@ def main(argv: list[str]) -> int:
     )
     args = ap.parse_args(argv)
 
-    if args.self_check:
-        import subprocess
-
-        here = pathlib.Path(__file__).resolve().parent
-        diff_text = subprocess.run(
-            ["git", "diff", "--", "scripts/anticheat_scan.py",
-             "scripts/test_anticheat_scan.py"],
-            cwd=here.parent, capture_output=True, text=True, check=True,
-        ).stdout
-    elif args.diff:
-        with open(args.diff, encoding="utf-8") as fh:
-            diff_text = fh.read()
-    elif not sys.stdin.isatty():
-        diff_text = sys.stdin.read()
-    else:
-        diff_text = ""
-
-    files = args.files.split(",") if args.files else None
-    trajectory = None
-    if args.trajectory:
-        with open(args.trajectory, encoding="utf-8") as fh:
-            trajectory = json.load(fh)
+    # Gathering inputs is where a *broken invocation* surfaces (unreadable file,
+    # malformed JSON, failed git). Fail with the distinct operational code so it
+    # is never confused with a clean scan or a findings downgrade.
+    try:
+        if args.self_check:
+            here = pathlib.Path(__file__).resolve().parent
+            diff_text = subprocess.run(
+                ["git", "diff", "--", "scripts/anticheat_scan.py",
+                 "scripts/test_anticheat_scan.py"],
+                cwd=here.parent, capture_output=True, text=True, check=True,
+            ).stdout
+        elif args.diff:
+            with open(args.diff, encoding="utf-8") as fh:
+                diff_text = fh.read()
+        elif not sys.stdin.isatty():
+            diff_text = sys.stdin.read()
+        else:
+            diff_text = ""
+
+        files = args.files.split(",") if args.files else None
+        trajectory = None
+        if args.trajectory:
+            with open(args.trajectory, encoding="utf-8") as fh:
+                trajectory = json.load(fh)
+    except (OSError, json.JSONDecodeError, subprocess.CalledProcessError) as exc:
+        print(f"{ap.prog}: operational error: {exc}", file=sys.stderr)
+        return EXIT_OPERATIONAL_ERROR
 
     result = scan(diff_text=diff_text, changed_files=files, trajectory=trajectory)
     print(json.dumps(result, indent=2))
     if result["downgrade_to"] == "FailedSafety":
-        return 2
-    return 0 if result["clean"] else 1
+        return EXIT_GATE_TAMPERING
+    return EXIT_CLEAN if result["clean"] else EXIT_FINDINGS
 
 
 if __name__ == "__main__":

scripts/rollout_ledger.py

@@ -36,6 +36,10 @@
 import sys
 from pathlib import Path
 
+
+def _warn(message: str) -> None:
+    print(f"rollout_ledger: {message}", file=sys.stderr)
+
 RECORD_FIELDS = (
     "id",
     "parent",
@@ -64,28 +68,57 @@ def append(record: dict, path: str | Path) -> dict:
     return written
 
 
-def read(path: str | Path) -> list[dict]:
-    """Return every record in the ledger, in append order. Empty if absent."""
+def _read_with_stats(path: str | Path) -> tuple[list[dict], int]:
+    """Read the ledger, tolerating corruption. Returns ``(records, malformed)``.
+
+    A ledger is append-only JSONL that may have been truncated mid-write or hand-
+    edited, so a single bad line must not lose the whole lineage. A line that is
+    not a JSON object (unparseable, or valid JSON that is not a dict) is skipped
+    with a stderr warning and counted, never raised.
+    """
     p = Path(path)
     if not p.exists():
-        return []
-    records = []
-    for line in p.read_text(encoding="utf-8").splitlines():
-        if line.strip():
-            records.append(json.loads(line))
+        return [], 0
+    records: list[dict] = []
+    malformed = 0
+    for lineno, line in enumerate(p.read_text(encoding="utf-8").splitlines(), start=1):
+        if not line.strip():
+            continue
+        try:
+            record = json.loads(line)
+        except json.JSONDecodeError as exc:
+            malformed += 1
+            _warn(f"skipping malformed ledger line {lineno}: {exc}")
+            continue
+        if not isinstance(record, dict):
+            malformed += 1
+            _warn(f"skipping malformed ledger line {lineno}: not a JSON object")
+            continue
+        records.append(record)
+    return records, malformed
+
+
+def read(path: str | Path) -> list[dict]:
+    """Return every valid record in the ledger, in append order. Empty if absent.
+
+    Malformed lines are skipped (with a stderr warning), never fatal.
+    """
+    records, _ = _read_with_stats(path)
     return records
 
 
 def summarize(path: str | Path) -> dict:
-    """Compute repair-productivity (productive fraction) and the candidate count."""
-    records = read(path)
+    """Compute repair-productivity (productive fraction), the candidate count, and
+    the number of malformed lines skipped."""
+    records, malformed = _read_with_stats(path)
     count = len(records)
     productive = sum(1 for r in records if r.get("productive"))
     repair_productivity = productive / count if count else 0.0
     return {
         "count": count,
         "productive": productive,
         "repair_productivity": repair_productivity,
+        "malformed": malformed,
     }
 
 

scripts/runtime_monitor.py

@@ -287,14 +287,30 @@ def health_report(loop_dir) -> dict:
     }
 
 
+def _exit_code(report: dict) -> int:
+    """Map a health report to an explicit exit code so a script/CI can react
+    without re-parsing the JSON:
+
+      * 0 — healthy or already terminal: recommendation ``continue``/``done``.
+      * 1 — intervention recommended (``replan``/``revert``/``approval``) or a
+            ``degraded`` RUNLOG whose detectors are inert.
+      * 2 — precondition/operational error (missing or invalid loop state).
+    """
+    if report.get("status") == "error":
+        return 2
+    if report.get("recommendation") in {"continue", "done"}:
+        return 0
+    return 1
+
+
 def main(argv: list[str] | None = None) -> int:
     argv = list(sys.argv[1:] if argv is None else argv)
     if not argv:
         print("usage: runtime_monitor.py <path-to-.loop-dir>", file=sys.stderr)
         return 2
     report = health_report(argv[0])
     print(json.dumps(report, indent=2))
-    return 0
+    return _exit_code(report)
 
 
 if __name__ == "__main__":