"""AZ.1 — JSON serialization for the semantic lock dict.
Phase AZ replaces the byte-locked seeds (~28 MB across 3 dialects)
with per-(instance, dialect) JSON semantic locks gated on the
violation SET rather than SQL bytes. This module is the
serialization seam: `semantic_lock(conn, ALL_INVARIANTS)` returns
`dict[str, frozenset[Violation]]`; `lock_to_json` renders the
canonical JSON shape AZ.0 designed.
Per AZ.0's "JSON-string equality is the gate contract" — the
loader doesn't reconstruct `Violation` objects. The test
serializes the live emit through `lock_to_json` + compares the
output string against the on-disk file byte-for-byte.
See `docs/audits/az_0_semantic_lock_schema.md` for the full
design + the validation table showing the semantic lock catches
every real violation-set change AY.5 caught + drops every
byte-only false positive.
"""
from __future__ import annotations
import json
from datetime import date
from typing import Any
from recon_gen.common.spine.violation import (
AuditFixture,
CoverageObservation,
RuleViolation,
Violation,
)
from recon_gen.common.sql import Dialect
_SCHEMA_VERSION = 1
"""Per AZ.0 — bumps when the JSON serialization shape changes (a
`dict` vs `list` flip on `identity`, or a new mandatory top-level
key). Loaders gate on version match. AZ.6's release notes pin
the bump-policy doc."""
_COMMENT_TEXT = (
"Auto-generated by `recon-gen data semantic-lock`. Hand-edits "
"will be overwritten on next re-lock. Drift here means the "
"spine emit produced a different violation set than the locked "
"snapshot — investigate before re-locking."
)
def _kind_for(v: Violation) -> str:
"""Map the AY.2.a Violation subtype to its JSON `kind` string.
Pattern-matches on concrete subclass for precision; falls back
to the abstract `Violation` base with a loud error so a new
subtype lands without silently serializing as the wrong kind."""
if isinstance(v, RuleViolation):
return "rule_violation"
if isinstance(v, CoverageObservation):
return "coverage"
if isinstance(v, AuditFixture):
return "audit_fixture"
raise TypeError(
f"semantic_lock_json: unknown Violation subtype "
f"{type(v).__name__!r} — add a `kind` mapping in `_kind_for` "
f"when promoting a new subtype on the spine."
)
def _value_to_json(value: object) -> Any:
"""Normalize an identity-tuple value to a JSON-native type.
JSON natively handles `str` / `int` / `float` / `bool` / `None`.
Per AZ.0: `date` → ISO-8601 string (`"2030-01-02"`).
Defensive fallback for unanticipated types: `repr(value)` (so
the lock still serializes, but the loader gets a hint that a
new identity-value type landed without explicit handling here).
Tests should catch this at re-lock time.
"""
if value is None or isinstance(value, (str, int, float, bool)):
return value
if isinstance(value, date):
return value.isoformat()
return repr(value) # defensive — opaque but byte-stable
[docs]
def lock_to_json(
lock: dict[str, frozenset[Violation]],
*,
instance: str,
dialect: Dialect,
canonical_anchor: date,
) -> str:
"""Render a semantic lock dict to the canonical JSON shape.
Deterministic + byte-stable across runs: invariant names
sorted alphabetically; violations within each invariant sorted
by `(kind, sorted-identity-repr)`; identity keys sorted
alphabetically per entry.
Empty invariants land as empty arrays (not omitted) so diffs
surface "X used to fire, doesn't now" clearly.
Args:
lock: `dict[invariant_name, frozenset[Violation]]` — exactly
what `semantic_lock(conn, ALL_INVARIANTS)` returns.
instance: the L2 instance name (e.g., "spec_example") —
lands in `scenario_fingerprint.instance` for mismatch
detection.
dialect: the SQL dialect — lands in
`scenario_fingerprint.dialect`.
canonical_anchor: the canonical lock date — lands in
`scenario_fingerprint.canonical_anchor`.
Returns:
A JSON string with `indent=2`, terminating newline. Suitable
for direct write-to-disk + byte-equality test.
"""
violations_json: dict[str, list[dict[str, Any]]] = {}
for invariant_name in sorted(lock):
entries: list[dict[str, Any]] = []
for v in lock[invariant_name]:
identity = {
str(k): _value_to_json(val)
for k, val in sorted(v.identity, key=lambda kv: str(kv[0]))
}
entries.append({
"kind": _kind_for(v),
"identity": identity,
})
# Sort entries by (kind, identity-tuple-sorted-repr) so re-runs
# are byte-stable regardless of frozenset iteration order.
entries.sort(key=lambda e: (
str(e["kind"]),
tuple(sorted(e["identity"].items())),
))
violations_json[invariant_name] = entries
payload = {
"_comment": _COMMENT_TEXT,
"scenario_fingerprint": {
"instance": instance,
"dialect": dialect.value,
"canonical_anchor": canonical_anchor.isoformat(),
"schema_version": _SCHEMA_VERSION,
},
"violations": violations_json,
}
# `sort_keys=False` — explicit per-level sort done above; sort_keys
# would re-order `scenario_fingerprint` fields away from their
# natural (instance → dialect → anchor → version) reading order.
return json.dumps(payload, indent=2, sort_keys=False) + "\n"