Gate promotion on lint errors

README.md

@@ -69,12 +69,30 @@ Lint the import:
 groundrecall lint .groundrecall/imports/<import-id>
 ```
 
+Review significant imports:
+
+```bash
+groundrecall review-server .groundrecall/imports/<import-id>
+```
+
 Promote the import into a canonical store:
 
 ```bash
 groundrecall promote .groundrecall/imports/<import-id> .groundrecall/store --reviewer your-name
 ```
 
+Promotion refuses imports with lint errors by default. Fix the source material,
+adapter, or review data first. If you intentionally need to preserve a flawed
+import for triage or recovery, use:
+
+```bash
+groundrecall promote .groundrecall/imports/<import-id> .groundrecall/store \
+  --reviewer your-name \
+  --allow-lint-errors
+```
+
+Warnings remain visible in the review queue but do not block promotion.
+
 Inspect or query the store:
 
 ```bash

docs/quickstart.md

@@ -56,12 +56,26 @@ Inspect the import outputs:
 groundrecall lint imports/<import-id>
 ```
 
+For anything non-trivial, open the review bundle before promotion:
+
+```bash
+groundrecall review-server imports/<import-id>
+```
+
 Promote the imported review artifacts into a canonical store:
 
 ```bash
 groundrecall promote imports/<import-id> store/
 ```
 
+Promotion is gated by lint errors. Warnings are retained for review, but errors
+must be repaired before promotion unless you explicitly choose to keep the
+import as triage material:
+
+```bash
+groundrecall promote imports/<import-id> store/ --allow-lint-errors
+```
+
 ## Query The Canonical Store
 
 Query a concept:

src/groundrecall/__init__.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from .inspect import inspect_store, summarize_store
 from .ingest import ImportResult, build_parser as build_import_parser, main as import_main, run_groundrecall_import
 from .models import *  # noqa: F403
-from .promotion import build_parser as build_promotion_parser, main as promotion_main, promote_import_to_store
+from .promotion import PromotionGateError, build_parser as build_promotion_parser, main as promotion_main, promote_import_to_store
 from .query import (
     build_parser as build_query_parser,
     build_query_bundle_for_concept,
@@ -22,6 +22,7 @@ __all__ = [
     "build_import_parser",
     "import_main",
     "promote_import_to_store",
+    "PromotionGateError",
     "build_promotion_parser",
     "promotion_main",
     "query_concept",

src/groundrecall/promotion.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import sys
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
@@ -20,6 +21,14 @@ from .review_schema import ReviewSession
 from .store import GroundRecallStore
 
 
+class PromotionGateError(RuntimeError):
+    """Raised when an import is not eligible for promotion."""
+
+    def __init__(self, message: str, payload: dict[str, Any]) -> None:
+        super().__init__(message)
+        self.payload = payload
+
+
 def _read_json(path: Path) -> dict[str, Any]:
     return json.loads(path.read_text(encoding="utf-8"))
 
@@ -76,13 +85,50 @@ def _review_candidate_rationale(item: dict[str, Any]) -> str:
     return " | ".join(parts)
 
 
+def _load_lint_payload(import_dir: Path) -> dict[str, Any]:
+    lint_path = import_dir / "lint_findings.json"
+    if not lint_path.exists():
+        return {
+            "summary": {"error_count": 0, "warning_count": 0},
+            "findings": [],
+            "missing_lint_file": True,
+        }
+    payload = _read_json(lint_path)
+    payload.setdefault("summary", {})
+    payload.setdefault("findings", [])
+    return payload
+
+
+def _lint_error_payload(lint_payload: dict[str, Any]) -> dict[str, Any]:
+    findings = [item for item in lint_payload.get("findings", []) if item.get("severity") == "error"]
+    return {
+        "error_count": int(lint_payload.get("summary", {}).get("error_count", len(findings))),
+        "warning_count": int(lint_payload.get("summary", {}).get("warning_count", 0)),
+        "errors": findings,
+    }
+
+
+def _enforce_promotion_gate(import_dir: Path, allow_lint_errors: bool) -> dict[str, Any]:
+    lint_payload = _load_lint_payload(import_dir)
+    gate_payload = _lint_error_payload(lint_payload)
+    if gate_payload["error_count"] > 0 and not allow_lint_errors:
+        raise PromotionGateError(
+            "Import has lint errors; review or repair the import before promotion, "
+            "or pass allow_lint_errors=True / --allow-lint-errors to override.",
+            gate_payload,
+        )
+    return gate_payload
+
+
 def promote_import_to_store(
     import_dir: str | Path,
     store_dir: str | Path,
     reviewer: str | None = None,
     snapshot_id: str | None = None,
+    allow_lint_errors: bool = False,
 ) -> dict[str, Any]:
     base = Path(import_dir)
+    gate_payload = _enforce_promotion_gate(base, allow_lint_errors=allow_lint_errors)
     manifest = _read_json(base / "manifest.json")
     review_session = ReviewSession.model_validate_json((base / "review_session.json").read_text(encoding="utf-8"))
     queue_payload = _read_json(base / "review_queue.json")
@@ -244,6 +290,9 @@ def promote_import_to_store(
         "promoted_concept_count": len(promoted_concept_ids),
         "promoted_claim_count": len(promoted_claim_ids),
         "promoted_relation_count": len(promoted_relation_ids),
+        "lint_error_count": gate_payload["error_count"],
+        "lint_warning_count": gate_payload["warning_count"],
+        "lint_errors_allowed": allow_lint_errors,
         "snapshot_id": built_snapshot.snapshot_id,
     }
 
@@ -254,15 +303,25 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("store_dir")
     parser.add_argument("--reviewer", default=None)
     parser.add_argument("--snapshot-id", default=None)
+    parser.add_argument(
+        "--allow-lint-errors",
+        action="store_true",
+        help="Promote even when lint_findings.json contains errors. Warnings do not block promotion.",
+    )
     return parser
 
 
 def main() -> None:
     args = build_parser().parse_args()
+    try:
         payload = promote_import_to_store(
             import_dir=args.import_dir,
             store_dir=args.store_dir,
             reviewer=args.reviewer,
             snapshot_id=args.snapshot_id,
+            allow_lint_errors=args.allow_lint_errors,
         )
+    except PromotionGateError as exc:
+        print(json.dumps({"ok": False, "error": str(exc), "gate": exc.payload}, indent=2), file=sys.stderr)
+        raise SystemExit(2) from exc
     print(json.dumps(payload, indent=2))

src/groundrecall/review_server.py

@@ -8,7 +8,7 @@ from pathlib import Path
 from urllib.parse import parse_qs, urlparse
 
 from .citation_support import materialize_citegeist_store
-from .promotion import promote_import_to_store
+from .promotion import PromotionGateError, promote_import_to_store
 from .review_workspace import GroundRecallReviewWorkspace
 
 
@@ -177,12 +177,17 @@ class GroundRecallReviewHandler(BaseHTTPRequestHandler):
             if not store_dir:
                 _json_response(self, 400, {"ok": False, "error": "store_dir is required"})
                 return
+            try:
                 result = promote_import_to_store(
                     import_dir=self.workspace.import_dir,
                     store_dir=store_dir,
                     reviewer=payload.get("reviewer"),
                     snapshot_id=payload.get("snapshot_id"),
+                    allow_lint_errors=bool(payload.get("allow_lint_errors")),
                 )
+            except PromotionGateError as exc:
+                _json_response(self, 409, {"ok": False, "error": str(exc), "gate": exc.payload})
+                return
             _json_response(self, 200, {"ok": True, "promotion": result})
             return
         if parsed.path == "/api/citations/verify":

tests/test_groundrecall_promotion.py

@@ -3,8 +3,10 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
+import pytest
+
 from groundrecall.ingest import run_groundrecall_import
-from groundrecall.promotion import promote_import_to_store
+from groundrecall.promotion import PromotionGateError, promote_import_to_store
 from groundrecall.store import GroundRecallStore
 
 
@@ -121,3 +123,54 @@ def test_groundrecall_promotion_preserves_queue_graph_rationale(tmp_path: Path)
     assert "bridge_concept" in bridge_candidate.finding_codes
     assert "lane=conflict_resolution" in bridge_candidate.rationale
     assert "graph=bridge_concept" in bridge_candidate.rationale
+
+
+def test_groundrecall_promotion_blocks_lint_errors_by_default(tmp_path: Path) -> None:
+    root = tmp_path / "llmwiki"
+    (root / "wiki").mkdir(parents=True)
+    (root / "wiki" / "bad.md").write_text("# Bad\n\n- A claim with a synthetic lint error.\n", encoding="utf-8")
+
+    result = run_groundrecall_import(root, mode="quick", import_id="lint-gate-test")
+    lint_path = result.out_dir / "lint_findings.json"
+    lint_payload = json.loads(lint_path.read_text(encoding="utf-8"))
+    lint_payload["summary"]["error_count"] = 1
+    lint_payload["findings"].append(
+        {
+            "severity": "error",
+            "code": "claim_missing_observation",
+            "target_id": "clm_synthetic",
+            "message": "Synthetic test error.",
+        }
+    )
+    lint_path.write_text(json.dumps(lint_payload, indent=2), encoding="utf-8")
+
+    with pytest.raises(PromotionGateError) as excinfo:
+        promote_import_to_store(result.out_dir, tmp_path / "store", reviewer="R")
+
+    assert excinfo.value.payload["error_count"] == 1
+    assert excinfo.value.payload["errors"][0]["code"] == "claim_missing_observation"
+
+
+def test_groundrecall_promotion_can_override_lint_error_gate(tmp_path: Path) -> None:
+    root = tmp_path / "llmwiki"
+    (root / "wiki").mkdir(parents=True)
+    (root / "wiki" / "override.md").write_text("# Override\n\n- A claim with an allowed lint error.\n", encoding="utf-8")
+
+    result = run_groundrecall_import(root, mode="quick", import_id="lint-gate-override-test")
+    lint_path = result.out_dir / "lint_findings.json"
+    lint_payload = json.loads(lint_path.read_text(encoding="utf-8"))
+    lint_payload["summary"]["error_count"] = 1
+    lint_payload["findings"].append(
+        {
+            "severity": "error",
+            "code": "relation_missing_source",
+            "target_id": "rel_synthetic",
+            "message": "Synthetic test error.",
+        }
+    )
+    lint_path.write_text(json.dumps(lint_payload, indent=2), encoding="utf-8")
+
+    payload = promote_import_to_store(result.out_dir, tmp_path / "store", reviewer="R", allow_lint_errors=True)
+
+    assert payload["lint_error_count"] == 1
+    assert payload["lint_errors_allowed"] is True