feat(tools): make_json_schema_validator — validate a parsed object against a JSON Schema

The generic base for the "validate against an official, vendored, version-pinned upstream
JSON schema" pattern: a consumer vendors the schema file + a refresh script and layers its
own bespoke checks on top; the kit owns only the validator wiring. Given a JSON Schema (dict
or path), returns a function that maps a parsed object to a list of violation messages
("<path>: <reason>", truncated at max_errors so a huge invalid doc can't flood the trace).
Parsing stays the consumer's job, so it composes with any extract/parse step.

`jsonschema` is an OPTIONAL dependency (rlm-kit[jsonschema]), imported lazily so `import
rlm_kit` and the dspy-free tools package stay lean. Consumer-driven: a downstream consumer was
hand-rolling structural gates that drift from the real upstream format; this is the reusable
half of moving to authoritative schema validation.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: qazbnm456

CHANGELOG.md

@@ -12,6 +12,16 @@ surfaced by dogfooding a real downstream consumer.
 
 ### Added
 
+- **`make_json_schema_validator` — validate a parsed object against a JSON Schema (draft 2020-12).**
+  The generic base for the "validate against an official, vendored, version-pinned upstream JSON
+  schema" pattern: a consumer vendors the schema file + a refresh script (the provider-specific half)
+  and layers its own bespoke checks on top; the kit owns only the validator wiring. Returns the
+  violation messages (`"<path>: <reason>"`, truncated at `max_errors` so a huge invalid doc can't
+  flood the trace) for a parsed dict — parsing stays the consumer's job, so it composes with any
+  extract/parse step. `jsonschema` is an OPTIONAL dependency (`rlm-kit[jsonschema]`), imported lazily
+  so `import rlm_kit` and the dspy-free `tools` package stay lean. Consumer-driven: a downstream
+  consumer was hand-rolling structural gates that drift from the real upstream format; this is the
+  reusable half of moving to authoritative schema validation.
 - **`get_sub_lm` promoted to the public surface** (`rlm_kit.get_sub_lm`; lazy re-export, keeps
   `import rlm_kit` dspy-free). Returns the base sub-LM `configure` built — the instance a consumer
   wraps with `intercept_sub_lm` before passing as `RLMTask(sub_lm=...)`. Consumer-driven: TWO

README.md

@@ -67,7 +67,7 @@ Deno sandbox (`brew install deno`) — the logic and tests run without either.
 | `task.py` | `RLMTask` base class. |
 | `_retry.py` | Validation + retry engine (dspy-free, unit-tested). |
 | `sandbox.py` | Interpreter selection + the insecure-sandbox guard. |
-| `tools/` | `make_schema_validator`, SSRF-guarded `make_fetch_tool`, provider-agnostic `make_web_search_tool`, and `make_model_tool` — the generic "model-as-tool + transient-retry + validate" core (a project wraps it with its own endpoint/validator/messages). |
+| `tools/` | `make_schema_validator` (pydantic) + `make_json_schema_validator` (validate a parsed object against a vendored JSON Schema — the base for the "validate against an official, version-pinned upstream schema" pattern; needs `rlm-kit[jsonschema]`), SSRF-guarded `make_fetch_tool`, provider-agnostic `make_web_search_tool`, and `make_model_tool` — the generic "model-as-tool + transient-retry + validate" core (a project wraps it with its own endpoint/validator/messages). |
 | `optimize.py` | GEPA harness — metric templates now, compile in Phase 2. |
 | `sub_lm.py` | `intercept_sub_lm` — wrap the RLM's sub-LM to trace every escalation as a `sub_call` (+ optional validate/post-process); `model_as_tool` for LM-decided multi-model routing. |
 | `skills.py` | `load_skills_as_tools` — expose a Skills directory to the RLM as tools. |

pyproject.toml

@@ -42,6 +42,11 @@ observe = [
 mcp = [
     "mcp>=1.0",
 ]
+# JSON-Schema validation (make_json_schema_validator): validate a parsed object against a
+# vendored, version-pinned upstream JSON schema. Opt-in — the core scaffold runs without it.
+jsonschema = [
+    "jsonschema>=4.0",
+]
 
 [project.urls]
 Homepage = "https://github.com/qazbnm456/rlm-kit"

rlm_kit/tools/__init__.py

@@ -8,10 +8,11 @@
 )
 from .model import ModelToolResult, make_model_tool
 from .search import make_web_search_tool, normalise_search_results
-from .validation import make_schema_validator
+from .validation import make_json_schema_validator, make_schema_validator
 
 __all__ = [
     "make_schema_validator",
+    "make_json_schema_validator",
     "is_safe_url",
     "resolved_host_is_safe",
     "parse_cidrs",

rlm_kit/tools/validation.py

@@ -1,17 +1,73 @@
-"""Schema-validation tool factory.
+"""Schema-validation tool factories.
 
-Produces a plain callable the RLM can invoke inside the REPL to check its draft
-JSON against the expected pydantic schema before emitting a final answer — the
-generalised form of the original app's ``validate_vulnerability_report``.
+Two shapes, both consumer-facing base primitives:
+
+- ``make_schema_validator(model)`` — a plain callable the RLM invokes inside the REPL to
+  check its draft JSON against a pydantic schema before emitting a final answer (returns a
+  human message). The generalised form of the original app's ``validate_vulnerability_report``.
+- ``make_json_schema_validator(schema)`` — validate a PARSED object against a JSON Schema
+  (draft 2020-12) and return the violation messages, the generic base for the "validate
+  against a vendored, version-pinned upstream JSON schema" pattern (a consumer vendors the
+  schema + a refresh script; the kit owns the validator wiring).
 """
 
 from __future__ import annotations
 
-from typing import Callable, Type
+import json
+import os
+from typing import Callable, Type, Union
 
 from pydantic import BaseModel
 
 
+def make_json_schema_validator(
+    schema: Union[dict, str, os.PathLike],
+    *,
+    max_errors: int = 20,
+) -> Callable[[object], list[str]]:
+    """Return a function that validates a PARSED object against a JSON Schema and returns a
+    list of human-readable violation messages (empty list = valid).
+
+    ``schema`` is the JSON Schema as a dict, or a path to a ``.json`` schema file (read once,
+    at factory time). The returned validator takes an already-parsed object (a dict from
+    ``yaml.safe_load`` / ``json.loads``) — parsing is the consumer's job, so this composes with
+    whatever extract/parse step precedes it. Each message is ``"<json/pointer/path>: <reason>"``
+    (root violations use ``"(root)"``); the list is truncated to ``max_errors`` (a huge invalid
+    doc must not flood the trace). Deterministic ordering (by error path) so traces are stable.
+
+    This is the GENERIC base for the "validate against an official, vendored, version-pinned
+    upstream JSON schema" pattern: a consumer vendors the schema file + a refresh script (the
+    provider-specific half), and layers its own bespoke checks on top; the kit owns only this
+    wiring. ``jsonschema`` is an OPTIONAL dependency (``rlm-kit[jsonschema]``) — imported lazily
+    so ``import rlm_kit`` and the dspy-free ``tools`` package stay lean.
+    """
+    try:
+        from jsonschema import Draft202012Validator
+    except ImportError as exc:  # pragma: no cover - exercised only without the extra installed
+        raise ImportError(
+            "make_json_schema_validator needs the optional 'jsonschema' dependency. "
+            "Install it with:  pip install 'rlm-kit[jsonschema]'"
+        ) from exc
+
+    if isinstance(schema, (str, os.PathLike)):
+        with open(schema, encoding="utf-8") as fh:
+            schema = json.load(fh)
+    validator = Draft202012Validator(schema)
+
+    def validate(obj: object) -> list[str]:
+        """Validate a parsed object against the JSON schema; return violation messages ([] = ok)."""
+        errors: list[str] = []
+        for err in sorted(validator.iter_errors(obj), key=lambda e: list(e.absolute_path)):
+            loc = "/".join(str(p) for p in err.absolute_path) or "(root)"
+            errors.append(f"{loc}: {err.message}")
+            if len(errors) >= max_errors:
+                errors.append(f"… (schema errors truncated at {max_errors})")
+                break
+        return errors
+
+    return validate
+
+
 def make_schema_validator(model: Type[BaseModel]) -> Callable[[str], str]:
     """Return a tool that validates a JSON string against ``model``.
 

tests/test_tools.py

@@ -14,7 +14,7 @@
 )
 from rlm_kit.tools.model import ModelToolResult, make_model_tool
 from rlm_kit.tools.search import make_web_search_tool, normalise_search_results
-from rlm_kit.tools.validation import make_schema_validator
+from rlm_kit.tools.validation import make_json_schema_validator, make_schema_validator
 from rlm_kit.trace import EVENT_TOOL_CALL, TraceRecorder, load_events
 
 
@@ -144,6 +144,46 @@ def test_schema_validator_reports_failure():
     assert "failed" in v('{"title": "t"}').lower()
 
 
+# ---- JSON-schema validator (make_json_schema_validator) -------------------
+
+_JSON_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "id": {"type": "string"},
+        "severity": {"enum": ["info", "low", "medium", "high", "critical"]},
+    },
+    "required": ["id", "severity"],
+}
+
+
+def test_json_schema_validator_passes_a_valid_object():
+    v = make_json_schema_validator(_JSON_SCHEMA)
+    assert v({"id": "x", "severity": "high"}) == []          # [] == valid
+
+
+def test_json_schema_validator_reports_each_violation_with_a_path():
+    v = make_json_schema_validator(_JSON_SCHEMA)
+    errs = v({"severity": "spicy"})                          # missing id + bad enum
+    assert any("id" in e for e in errs)                      # required-field violation
+    assert any("severity" in e and "spicy" in e for e in errs)  # located on the bad field
+
+
+def test_json_schema_validator_loads_schema_from_a_path(tmp_path):
+    import json as _json
+    p = tmp_path / "schema.json"
+    p.write_text(_json.dumps(_JSON_SCHEMA))
+    v = make_json_schema_validator(str(p))
+    assert v({"id": "x", "severity": "high"}) == []
+    assert v({"id": 1, "severity": "high"})                  # id wrong type → non-empty
+
+
+def test_json_schema_validator_truncates_a_flood_of_errors():
+    schema = {"type": "object", "additionalProperties": {"type": "string"}}
+    v = make_json_schema_validator(schema, max_errors=3)
+    errs = v({f"k{i}": i for i in range(50)})                # 50 int values → 50 violations
+    assert len(errs) == 4 and "truncated" in errs[-1]        # 3 + the truncation marker
+
+
 # ---- SSRF guard ----------------------------------------------------------
 
 @pytest.mark.parametrize(