spanforge.validate

JSON Schema validation for Event envelopes.

Validates Event instances against the published JSON Schema. Schema version is selected automatically from the event's schema_version field:

• "1.0" → schemas/v1.0/schema.json
• "2.0" (default) → schemas/v2.0/schema.json

When the optional jsonschema package is installed, full Draft 2020-12 validation is performed. Otherwise a stdlib-only structural check covers all required fields, types, and regex patterns.

Install for full validation:

pip install "spanforge[jsonschema]"

Module-level functions

validate_event(event: Event) -> None

Validate event against the published JSON Schema.

The schema version is read from event.schema_version and the matching schema file is selected automatically (RFC §15.5).

Args:

Raises:

• SchemaValidationError — if the event does not conform to the envelope schema.
• FileNotFoundError — if the matching schema file is missing from the distribution.
• TypeError — if event is not an Event instance.

Example:

from spanforge import Event, EventType
from spanforge.validate import validate_event

event = Event(
    event_type=EventType.TRACE_SPAN_COMPLETED,
    source="llm-trace@0.3.1",
    payload={"span_name": "run", "status": "ok"},
)
validate_event(event)  # passes silently

load_schema(version: Optional[str] = None) -> Dict[str, Any]

Load and cache a JSON Schema from disk by version.

The schema is loaded once per version key and cached in memory. If version is None, the current default ("2.0") is used. Unknown versions raise ValueError.

Args:

Returns: Dict[str, Any] — parsed JSON Schema as a plain Python dict.

Raises:

• FileNotFoundError — if the schema file cannot be found relative to the package root.
• ValueError — if an unknown schema version is requested.

Example:

from spanforge.validate import load_schema

schema_v2 = load_schema()        # loads schemas/v2.0/schema.json
schema_v1 = load_schema("1.0")   # loads schemas/v1.0/schema.json

Enforcement Modes (1.0.1)

EnforcementMode

Four validation enforcement levels:

ValidationResult

Dataclass returned by enforce_event():

enforce_event(event, mode=EnforcementMode.STRICT) -> ValidationResult

Validate and enforce an event according to the chosen mode.

from spanforge.validate import enforce_event, EnforcementMode

result = enforce_event(event, mode=EnforcementMode.WARN)
if result.violations:
    print(result.violations)

correct_event(doc: dict) -> dict

Correction pass that:

• Strips unknown top-level keys.
• Removes None-valued optional fields (trace_id, span_id, tags, checksum, signature).
• Normalises schema_version to the current default when the value is unrecognised.

Returns a new dict; does not mutate the input.

HMAC Signing (1.0.1)

sign_event_hmac(event: Event, key: str) -> Event

Sign an event with HMAC-SHA256. Returns a new Event with:

signature = "hmac-sha256:<64-hex-digest>"

The digest is computed over the canonical JSON representation of the event payload (sorted keys, no whitespace). Raises ValueError for an empty key.

from spanforge.validate import sign_event_hmac

signed = sign_event_hmac(event, key="my-secret-key")

Training Data Compliance Scanner

Current API (v1.0.1+): scan_dataset_compliance() in spanforge.sdk.dataset_scanner — EU AI Act Article 10 file-level scanner with HMAC-signed reports. See compliance API reference for full documentation.

Legacy row-level API (v1.0.0)

Deprecated. The row-level scan_dataset() API below is preserved for backwards compatibility. New integrations should use scan_dataset_compliance() instead.

scan_dataset(rows, *, check_pii_field_names=True, check_pii_values=True, required_fields=None) -> DatasetScanReport

Scan a list of record dicts (e.g. from a JSONL training dataset) for compliance issues:

• PII field names — flags fields whose names match known PII patterns (email, phone, ssn, passport, ip_address, biometric, gps, lat, lon, dob, national_id, driver_license, and more).
• PII values — flags field values that match email address, US phone number, or SSN patterns.
• Required field violations — flags records missing any field listed in required_fields.

from spanforge.validate import scan_dataset

rows = [
    {"prompt": "Hello", "email": "user@example.com"},
    {"prompt": "World"},
]
report = scan_dataset(rows, required_fields=["prompt", "response"])
print(report.total_findings)  # 2: pii_field_name + schema_violation

DatasetScanFinding

DatasetScanReport

Model Response Validation — spanforge.sdk.validate (1.0.1)

Distinct from spanforge.validate (event-envelope validation above). This module validates model responses on the hot path, running four ordered enforcement mechanisms before the response reaches the caller.

from spanforge.sdk import sf_validate          # singleton
from spanforge.sdk.validate import (
    SFValidateClient,
    ValidationResult,
    Violation,
    ValidateStatusInfo,
)

SFValidateClient

SFValidateClient(config: SFClientConfig, service_name: str = "sf-validate")

Four enforcement mechanisms run in order on every validate() call:

validate(response, *, schema=None, confidence_threshold=0.7, correction_fn=None, max_correction_passes=2, agent_id="", trace_id="") -> ValidationResult

Run all four enforcement mechanisms in order. Never raises on content violations — always returns ValidationResult.

Args:

Returns: ValidationResult

Example:

import re
from spanforge.sdk import sf_validate

# Schema check (regex)
result = sf_validate.validate(
    response="Order confirmed for customer@example.com",
    schema=r"^Order confirmed",
    agent_id="order-agent",
    trace_id="abc123",
)
print(result.passed)          # True or False
print(result.violations)      # list[Violation]
print(result.hmac_signature)  # HMAC of the audit record

# Multi-pass correction
def fix(resp, viols):
    return resp.replace("badword", "***")

result = sf_validate.validate(
    response="Please say badword now",
    correction_fn=fix,
)
print(result.correction_passes)    # 1
print(result.corrected_response)   # "Please say *** now"

get_status() -> ValidateStatusInfo

Return a snapshot of health and configuration:

info = sf_validate.get_status()
print(info.total_calls)            # total validate() calls
print(info.total_passed)           # calls where passed=True
print(info.jsonschema_available)   # True if jsonschema installed

ValidationResult dataclass

Returned by SFValidateClient.validate().

to_dict() -> dict — returns all fields as a plain dict.

Violation dataclass

Violation.__post_init__() raises ValueError for unknown severity values.

to_dict() -> dict — returns all fields as a plain dict.

ValidateStatusInfo dataclass

Audit schema keys

CLI — Dataset Scanner

# Scan a JSONL training dataset (exits 1 if any Article 10 clause fails)
spanforge validate --dataset training.jsonl

# Machine-readable JSON output
spanforge validate --dataset training.jsonl --output json

# PDF report (requires reportlab)
spanforge validate --dataset training.jsonl --output pdf