recon_gen.common.spine.invariant

Invariant Protocol — the rule + detector.

An Invariant owns its own name (matching the production matview suffix) and a detect() method that reads that matview’s output and projects each row to a Violation. The detection LOGIC stays in the matview SQL (common/l2/schema.py); detect() is a thin read of the already-computed rows, not a re-encoded copy of the detector.

What is NOT in the Protocol (intentionally):

  • scenario_for(…) — the smart constructor that resolves a shape selector to a ViolationGenerator. Each concrete invariant defines its own with the right kwargs (drift takes role; limit_breach takes (role, rail, direction); anomaly takes window + spike-magnitude). AS.2 promotes the concrete invariants with their per-shape scenario_for signatures.

That keeps the Protocol minimal — every L1 + L2 invariant can implement it without a kwargs-mismatch dance. The smart constructor varies by shape; the spine link (detect) does not.

Convention (de-facto across drift / overdraft / expected_eod / stuck_pending, formalized 2026-05-23): every concrete invariant’s scenario_for(…) accepts instance: L2Instance | None = None as a keyword-only argument. None defaults to the bundled spec_example yaml (the in-process test harness shape); callers thread an explicit L2Instance for production deploys and AT.x’s Investigation surface. NOT enforced by Protocol — convention only — pending the spike-before- locking decision on whether to formalize a from_instance(self, instance) -> Generator Protocol method (deferred until AU.4 + AT.2 give two more L2-coupling data points).

Promoted from tests/unit/test_as0_drift_full_spine.py by AS.1.

Classes

Invariant(*args, **kwargs)

A rule + detector.
class recon_gen.common.spine.invariant.Invariant(*args, **kwargs)[source]

Bases: Protocol

A rule + detector. Concrete invariants implement this via:

  1. A name ClassVar[str] matching the production matview suffix — "drift" reads from <prefix>_drift, etc. ClassVar here so concrete impls (frozen dataclasses) declare name: ClassVar[str] = "..." without ever shadowing it as an instance attribute. The Protocol-variance dance pyright cares about: ClassVar on both sides keeps the read-only contract honest.

  2. A detect(conn) method returning the breaches currently in the data, as a set[Violation].

runtime_checkable so isinstance(x, Invariant) works for the taxonomy bookkeeping in AS.2 — the invariant → {generators, views} map needs runtime lookup, not just static type checking.

detect(conn)[source]
Return type:

set[Violation]

Parameters:

conn (SyncConnection)

name: ClassVar[str]