Add lint

cli/ponens/registry.py

@@ -21,6 +21,7 @@
 from .formatting import (
     bold, gray, red, green, yellow, blue, cyan, magenta, underline, table, heading,
 )
+from .policy_compiler import CheckError, check_policy
 
 DEFAULT_GALLERY_URL = "https://ponens.dev/gallery/policies"
 
@@ -450,6 +451,62 @@ def cmd_policies_add(args):
     print(gray(f"  run: ponens trace check {path}"))
 
 
+def cmd_policies_lint(args):
+    """Lint policy definitions locally: required fields + formula syntax.
+
+    Runs the same oracle `trace check` applies before evaluating
+    (policy_compiler.check_policy), so a policy that lints valid cannot be
+    marked syntax-invalid by check later. With --json, exits 0 whenever
+    linting ran — per-policy verdicts are in the records; without it, exits 1
+    if any policy is invalid. An unreadable or malformed input file exits 1.
+    """
+    try:
+        with open(args.policy_file) as f:
+            policy_data = json.load(f)
+    except (OSError, json.JSONDecodeError) as e:
+        print(f"Error: cannot read policy file: {e}", file=sys.stderr)
+        return 1
+    policies = policy_data if isinstance(policy_data, list) else policy_data.get("policies", [])
+    if not isinstance(policies, list) or not all(isinstance(p, dict) for p in policies):
+        print('Error: policy file must be a JSON array of policy objects, or {"policies": [...]}',
+              file=sys.stderr)
+        return 1
+
+    as_json = getattr(args, "json", False)
+    records = []
+    invalid = 0
+    for p in policies:
+        pid = p.get("policy_id", p.get("name", "?"))
+        if "name" not in p:
+            # check_policy indexes policy['name'] before it can report it missing;
+            # produce the message it would have produced.
+            errors, warnings = [CheckError("Missing required field 'name'", pid)], []
+        else:
+            try:
+                _, errors, warnings = check_policy(p)
+            except Exception as e:  # a policy the checker cannot inspect is invalid, not a crash
+                errors, warnings = [CheckError(f"Policy could not be checked: {e!r}", pid)], []
+        record = {"policy_id": pid, "status": "invalid" if errors else "valid"}
+        if errors:
+            record["errors"] = [{"message": e.message, "path": e.path} for e in errors]
+            invalid += 1
+        if warnings:
+            record["warnings"] = [{"message": w.message, "path": w.path} for w in warnings]
+        records.append(record)
+        if not as_json:
+            print(f"  {'INVALID' if errors else 'OK     '} {pid}")
+            for e in errors:
+                print(f"          {e.message}")
+            for w in warnings:
+                print(f"          warning: {w.message}")
+
+    if as_json:
+        print(json.dumps(records, indent=2, ensure_ascii=False))
+        return 0
+    print(f"\n  {len(records)} policies linted, {invalid} invalid")
+    return 1 if invalid else 0
+
+
 # ----------------------------------------------------------------------------
 # Commands — unified search (policies, packs, organizations)
 # ----------------------------------------------------------------------------
@@ -611,3 +668,9 @@ def register(subparsers):
     p.add_argument("--into", required=True, help="Path to the trace JSON file")
     p.add_argument("--refresh", action="store_true", help="Force re-fetch, bypassing the cache")
     p.set_defaults(func=cmd_policies_add)
+
+    p = pol_sub.add_parser("lint", help="Lint policy definitions locally (required fields + formula syntax)")
+    p.add_argument("policy_file", help='Policy JSON file: an array of policies, or {"policies": [...]}')
+    p.add_argument("--json", action="store_true",
+                   help="Emit per-policy verdicts as JSON (machine-readable; exits 0 whenever linting ran)")
+    p.set_defaults(func=cmd_policies_lint)

cli/tests/unit/test_policies_lint.py

@@ -0,0 +1,94 @@
+"""`ponens policies lint` — local policy validation (required fields + formula
+syntax), the same oracle `trace check` applies before evaluating."""
+
+import json
+from argparse import Namespace
+
+from ponens.registry import cmd_policies_lint
+
+VALID = {
+    "policy_id": "tests_before_commit",
+    "name": "tests_before_commit",
+    "severity": "error",
+    "scope": "trace",
+    "kind": "temporal",
+    "formula": "G(GitCommit → P(RunTests ∧ completed))",
+}
+
+
+def _lint(tmp_path, payload, as_json=True):
+    f = tmp_path / "policies.json"
+    f.write_text(json.dumps(payload, ensure_ascii=False))
+    return cmd_policies_lint(Namespace(policy_file=str(f), json=as_json))
+
+
+def _records(capsys):
+    return json.loads(capsys.readouterr().out)
+
+
+def test_valid_policy_lints_valid(tmp_path, capsys):
+    assert _lint(tmp_path, [VALID]) == 0
+    (rec,) = _records(capsys)
+    assert rec == {"policy_id": "tests_before_commit", "status": "valid"}
+
+
+def test_wrapped_policies_key_is_accepted(tmp_path, capsys):
+    assert _lint(tmp_path, {"policies": [VALID]}) == 0
+    (rec,) = _records(capsys)
+    assert rec["status"] == "valid"
+
+
+def test_missing_scope_is_invalid_with_the_check_message(tmp_path, capsys):
+    p = {k: v for k, v in VALID.items() if k != "scope"}
+    assert _lint(tmp_path, [p]) == 0  # --json: verdicts in records, not the exit code
+    (rec,) = _records(capsys)
+    assert rec["status"] == "invalid"
+    assert any("Missing required field 'scope'" in e["message"] for e in rec["errors"])
+
+
+def test_formula_that_does_not_parse_is_invalid(tmp_path, capsys):
+    p = dict(VALID, formula="G(GitCommit →")
+    assert _lint(tmp_path, [p]) == 0
+    (rec,) = _records(capsys)
+    assert rec["status"] == "invalid"
+    assert rec["errors"]
+
+
+def test_missing_name_is_invalid_not_a_crash(tmp_path, capsys):
+    p = {k: v for k, v in VALID.items() if k != "name"}
+    assert _lint(tmp_path, [p]) == 0
+    (rec,) = _records(capsys)
+    assert rec["policy_id"] == "tests_before_commit"
+    assert rec["status"] == "invalid"
+    assert any("Missing required field 'name'" in e["message"] for e in rec["errors"])
+
+
+def test_human_mode_exits_nonzero_on_invalid(tmp_path, capsys):
+    p = {k: v for k, v in VALID.items() if k != "scope"}
+    assert _lint(tmp_path, [p], as_json=False) == 1
+    out = capsys.readouterr().out
+    assert "INVALID" in out and "1 invalid" in out
+
+
+def test_human_mode_exits_zero_when_all_valid(tmp_path, capsys):
+    assert _lint(tmp_path, [VALID], as_json=False) == 0
+    assert "0 invalid" in capsys.readouterr().out
+
+
+def test_malformed_file_is_refused(tmp_path, capsys):
+    f = tmp_path / "nope.json"
+    f.write_text("{not json")
+    assert cmd_policies_lint(Namespace(policy_file=str(f), json=True)) == 1
+    captured = capsys.readouterr()
+    assert captured.out == ""  # no records on refusal
+    assert "cannot read policy file" in captured.err
+
+
+def test_non_object_entries_are_refused(tmp_path, capsys):
+    assert _lint(tmp_path, ["just a string"]) == 1
+    assert "array of policy objects" in capsys.readouterr().err
+
+
+def test_empty_list_lints_clean(tmp_path, capsys):
+    assert _lint(tmp_path, []) == 0
+    assert _records(capsys) == []