open-standards/jouleclaw/jouleclaw-rs/crates/jouleclaw-hydra-client/clients/python/jclaw_hydra/shacl.py
Transaction Science 2032b91125 jouleclaw-hydra-client: Python + Go reference clients, byte-equal conformance (wave-23 #2)
Three independent implementations of the third-generation Hydra
client — Rust + Python + Go — all produce byte-for-byte identical
BLAKE3 input/output hashes against the same 9-vector conformance
suite. Cross-language conformance is what makes the Hydra-client
surface a real open standard rather than a Rust house format.

What ships:

  conformance/call-reports.json — 5 dispatch + 4 SHACL-validation
                                  vectors. The Rust producer is
                                  the oracle (regenerated via
                                  examples/gen_conformance_vectors.rs).

  clients/python/jclaw_hydra/   — Python reference client. Package
                                  layout for `from jclaw_hydra
                                  import HydraClient`. Uses the
                                  blake3 PyPI package for BLAKE3.
                                  Conformance test at
                                  clients/python/tests/test_conformance.py:
                                  9/9 PASS.

  clients/go/                   — Go reference client. Importable
                                  as `jclawhydra`. Uses
                                  lukechampine.com/blake3.
                                  Conformance via `go test ./...`:
                                  9/9 PASS across TestCallReportsConformance
                                  + TestValidationConformance.

  examples/gen_conformance_vectors.rs — Rust regenerator. Produces
                                        call-reports.json from the
                                        canonical Rust client; the
                                        other implementations
                                        verify against this output.

  conformance/README.md         — vector description, how to run
                                  the three implementations, scope.

The clients implement the same surface verbatim:

  - parse Hydra ApiDocumentation from JSON-LD
  - find operation by jc_tool_id / @id / title
  - validate input against the linked SHACL shape (open-world
    default if no shape registered; #Nothing-expects skips)
  - dispatch via Transport — echo-mode MockTransport ships as the
    test reference; transport returns { "echo": <input> } verbatim
    (matches Rust core; no "op" sidecar like the WASM demo wrapper)
  - canonical-JSON serialise the input + body (lexically-sorted
    keys, no whitespace) and BLAKE3-hash → hex
  - return CallReport with tool_id, status, input_hash, output_hash,
    status_matches_contract, body, touch{tool_id, joules_uj,
    energy_provenance}

The hashes match because the canonical-JSON form matches — that's
the spec primitive. The body match proves the MockTransport
agrees too. The status_matches_contract bit verifies the three
agree on how to walk an Operation's statusCodes menu.

Validation reports verify the same conforms-bit across the four
SHACL cases the conformance file declares: well-shaped input,
missing required field, bad enum, and open-world skip for an op
with no expects-shape.

Workspace: 107 native crates (+0; clients/go and clients/python
live outside Cargo), 2,635 lib tests pass, 0 failures.

Next wave-23 step: #3 — carry the vocab + SHACL pattern into the
four sister standards (ARL, OpenPay, Smart Byte, EOC, WAI) so the
whole TX-Science quintet speaks Gen-3 with stable IRIs.
2026-06-02 13:54:42 -04:00

208 lines
6.7 KiB
Python

"""Minimal SHACL Core validator — Python mirror of jouleclaw-shacl."""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
JC_NS = "https://jouleclaw.transaction.science/vocab/v1#"
class NodeKind(str, Enum):
IRI = "IRI"
LITERAL = "Literal"
BLANK_NODE_OR_IRI = "BlankNodeOrIRI"
class Datatype(str, Enum):
STRING = "string"
INTEGER = "integer"
NON_NEGATIVE_INTEGER = "non_negative_integer"
DOUBLE = "double"
BOOLEAN = "boolean"
DATETIME = "datetime"
@dataclass
class Violation:
shape: str
path: str
message: str
@dataclass
class ValidationReport:
conforms: bool
violations: List[Violation] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return {
"conforms": self.conforms,
"violations": [
{"shape": v.shape, "path": v.path, "message": v.message}
for v in self.violations
],
}
@dataclass
class PropertyShape:
path: str
min_count: Optional[int] = None
max_count: Optional[int] = None
datatype: Optional[Datatype] = None
node_kind: Optional[NodeKind] = None
in_values: Optional[List[Any]] = None
pattern: Optional[str] = None
min_inclusive: Optional[float] = None
max_inclusive: Optional[float] = None
node: Optional[str] = None
@dataclass
class NodeShape:
target_class: str
properties: List[PropertyShape] = field(default_factory=list)
class ShapeGraph:
def __init__(self) -> None:
self._shapes: Dict[str, NodeShape] = {}
def add(self, shape: NodeShape) -> "ShapeGraph":
self._shapes[shape.target_class] = shape
return self
def get(self, target_class: str) -> Optional[NodeShape]:
return self._shapes.get(target_class)
def __len__(self) -> int:
return len(self._shapes)
def validate(graph: ShapeGraph, target_class: str, value: Any) -> ValidationReport:
shape = graph.get(target_class)
if shape is None:
raise KeyError(f"no shape registered for {target_class}")
violations: List[Violation] = []
_validate_against(shape, value, graph, violations)
return ValidationReport(conforms=not violations, violations=violations)
def _validate_against(
shape: NodeShape,
value: Any,
graph: ShapeGraph,
out: List[Violation],
) -> None:
if not isinstance(value, dict):
out.append(Violation(shape.target_class, "", f"expected object, got {type(value).__name__}"))
return
for prop in shape.properties:
present = prop.path in value and value[prop.path] is not None
if not present:
if (prop.min_count or 0) > 0:
out.append(
Violation(
shape.target_class,
prop.path,
f"missing required property `{prop.path}`",
)
)
continue
_check_property(prop, value[prop.path], shape, graph, out)
def _check_property(
prop: PropertyShape,
value: Any,
parent: NodeShape,
graph: ShapeGraph,
out: List[Violation],
) -> None:
if isinstance(value, list):
n = len(value)
if prop.min_count is not None and n < prop.min_count:
out.append(_v(parent, prop, f"min_count {prop.min_count} not met ({n} present)"))
if prop.max_count is not None and n > prop.max_count:
out.append(_v(parent, prop, f"max_count {prop.max_count} exceeded ({n} present)"))
for item in value:
_check_value(prop, item, parent, graph, out)
return
_check_value(prop, value, parent, graph, out)
def _check_value(
prop: PropertyShape,
value: Any,
parent: NodeShape,
graph: ShapeGraph,
out: List[Violation],
) -> None:
# node_kind
if prop.node_kind is not None:
if prop.node_kind == NodeKind.IRI and not isinstance(value, str):
out.append(_v(parent, prop, f"node_kind IRI not satisfied"))
elif prop.node_kind == NodeKind.LITERAL and not isinstance(value, (str, int, float, bool)):
out.append(_v(parent, prop, f"node_kind Literal not satisfied"))
elif prop.node_kind == NodeKind.BLANK_NODE_OR_IRI and not isinstance(value, (dict, str)):
out.append(_v(parent, prop, f"node_kind BlankNodeOrIRI not satisfied"))
# datatype
if prop.datatype is not None and not _matches_datatype(prop.datatype, value):
out.append(_v(parent, prop, f"datatype {prop.datatype.value} not satisfied"))
# in
if prop.in_values is not None and value not in prop.in_values:
out.append(_v(parent, prop, f"value not in allowed set ({len(prop.in_values)} options)"))
# pattern
if prop.pattern is not None and isinstance(value, str):
if not re.search(prop.pattern, value):
out.append(_v(parent, prop, f"pattern `{prop.pattern}` not matched"))
# range
if isinstance(value, (int, float)) and not isinstance(value, bool):
f = float(value)
if prop.min_inclusive is not None and f < prop.min_inclusive:
out.append(_v(parent, prop, f"min_inclusive {prop.min_inclusive} not met ({f})"))
if prop.max_inclusive is not None and f > prop.max_inclusive:
out.append(_v(parent, prop, f"max_inclusive {prop.max_inclusive} exceeded ({f})"))
# node delegation
if prop.node is not None:
nested = graph.get(prop.node)
if nested is None:
out.append(_v(parent, prop, f"nested shape `{prop.node}` not in graph"))
else:
_validate_against(nested, value, graph, out)
def _matches_datatype(dt: Datatype, value: Any) -> bool:
if dt == Datatype.STRING:
return isinstance(value, str)
if dt == Datatype.INTEGER:
return isinstance(value, int) and not isinstance(value, bool)
if dt == Datatype.NON_NEGATIVE_INTEGER:
return isinstance(value, int) and not isinstance(value, bool) and value >= 0
if dt == Datatype.DOUBLE:
return isinstance(value, (int, float)) and not isinstance(value, bool)
if dt == Datatype.BOOLEAN:
return isinstance(value, bool)
if dt == Datatype.DATETIME:
if not isinstance(value, str) or len(value) < 19:
return False
return _looks_like_rfc3339(value)
return False
def _looks_like_rfc3339(s: str) -> bool:
# YYYY-MM-DDThh:mm:ss… the same shape jouleclaw-shacl accepts
return bool(
re.match(
r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}",
s,
)
)
def _v(parent: NodeShape, prop: PropertyShape, message: str) -> Violation:
return Violation(shape=parent.target_class, path=prop.path, message=message)