feat: consumer schema-location api and validate command

src/lokf/cli.py

@@ -2,6 +2,7 @@
 
     lokf new my-kb                                    # scaffold a knowledge base
     lokf convert path/to/concept.md --format ttl      # markdown -> RDF
+    lokf validate knowledge                           # assemble + validate a bundle
     lokf query examples/acme-knowledge "SELECT ..."   # SPARQL over a bundle
     lokf serve examples/acme-knowledge                # local SPARQL endpoint + viz
     lokf propose examples/acme-knowledge --apply      # typed relations from links
@@ -94,6 +95,62 @@ def convert(
         typer.echo(data, nl=False)
 
 
+# ---------------------------------------------------------------------------
+# validate
+# ---------------------------------------------------------------------------
+@app.command()
+def validate(
+    bundle_dir: Path = typer.Argument(
+        Path("knowledge"), exists=True, file_okay=False,
+        help="A LOKF bundle directory to assemble and validate.",
+    ),
+    schema: Optional[Path] = typer.Option(
+        None, "--schema", "-s",
+        help="Schema file (default: a local lokf.yaml checkout, else the copy "
+        "packaged with lokf).",
+    ),
+) -> None:
+    """Assemble a bundle and validate it against the LOKF schema.
+
+    Works on any bundle directory. The schema is resolved from ``--schema`` if
+    passed, otherwise from a ``lokf.yaml`` found in the current directory or an
+    ancestor, otherwise from the copy shipped inside the installed ``lokf``
+    package - so a bundle needs  no schema file of its own to be validated.
+    """
+    import os
+    import subprocess
+    import tempfile
+
+    from lokf.model import load_bundle
+    from lokf.schema import schema_path
+
+    try:
+        sch = schema_path(schema)
+    except FileNotFoundError as exc:
+        _err(str(exc))
+        raise typer.Exit(1)
+
+    bundle = load_bundle(bundle_dir)
+    doc = dict(bundle.meta)
+    doc["concepts"] = bundle.docs()
+
+    fd, tmp = tempfile.mkstemp(suffix=".bundle.json")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            json.dump(doc, f)
+        proc = subprocess.run(
+            ["linkml-validate", "-s", str(sch), "-C", "KnowledgeBundle", tmp]
+        )
+    finally:
+        os.remove(tmp)
+    if proc.returncode != 0:
+        raise typer.Exit(proc.returncode)
+    typer.echo(
+        f"OK — {len(bundle.concepts)} concepts in {bundle_dir} validate "
+        "against KnowledgeBundle."
+    )
+
+
 # ---------------------------------------------------------------------------
 # tables
 # ---------------------------------------------------------------------------

src/lokf/schema.py

@@ -67,6 +67,17 @@ def _load_context(resolved: str) -> dict:
     return json.loads(pathlib.Path(resolved).read_text(encoding="utf-8"))["@context"]
 
 
+def schema_path(path: str | pathlib.Path | None = None) -> pathlib.Path:
+    """Resolve the LOKF schema file (``lokf.yaml``) to a filesystem path.
+
+    Resolution order (same as every other reader here): an explicit *path*, a
+    ``lokf.yaml`` found in the current directory or an ancestor, the the copy
+    shipped inside the installed ``lokf`` package - so a bundle can be
+    validated without a schema file of its own.
+    """
+    return _resolve(_SCHEMA_NAME, path)
+
+
 def load_schema(path: str | pathlib.Path | None = None) -> dict:
     """Return the LOKF LinkML schema as a plain dict (cached per file)."""
     return _load_schema(str(_resolve(_SCHEMA_NAME, path)))

tests/test_cli.py

@@ -65,6 +65,20 @@ def test_convert_output_writes_file(tmp_path):
     assert "lokf:Metric" in text
 
 
+# -- validate ---------------------------------------------------------------
+def test_validate_reference_bundle_ok():
+    """validate <bundle> assembles and validates against KnowledgeBundle."""
+    result = runner.invoke(app, ["validate", str(BUNDLE)])
+    assert result.exit_code == 0
+    assert "validate against KnowledgeBundle" in result.stdout
+
+
+def test_validate_missing_bundle_nonzero_exit():
+    """A non-existent bundle directory is rejected before validation."""
+    result = runner.invoke(app, ["validate", "does-not-exist"])
+    assert result.exit_code != 0
+
+
 # -- query ------------------------------------------------------------------
 def test_query_select_table_contains_wau():
     """query <bundle> <SELECT> prints a table with the metric name."""