recon_gen.common.spine

The invariant spine (D6).

The typed source-of-truth for the project’s invariant-violation model: Violation is the currency that flows; Invariant.detect() produces them, ViolationGenerator.emit() manufactures the rows that should trip them. Promoted from the AP.3 + AS.0 spikes (docs/audits/ date_range_model_audit.md §5 “AS.0 result”).

Module layout (AS.1):

violation.py — Violation frozen dataclass + Violation.of() smart constructor. The currency type; no behaviour, just identity.
invariant.py — Invariant Protocol. name + detect(conn) -> set[Violation]. Detectors are thin SQL reads of the existing matview output, NOT re-encoded matview logic.
generator.py — ViolationGenerator Protocol. intended + emit (conn) -> None. Producer ≠ thing-produced; the generator claims to cause a violation, the violation is what detect() returns.

What is NOT here:

Concrete invariants / generators (DriftInvariant etc.) — AS.2 lands those alongside the per-invariant smart constructors (Invariant.scenario_for(…)) that vary per shape selector.
View — stays on common/tree/date_view.py; AR.1 already promoted it. The spine references it.
The stateful-fold base for ViolationGenerator (the AP.2 shape) — AS.3 lands it; AS.1’s Protocol is minimal so per-invariant generators can specialize freely.

class recon_gen.common.spine.AccountSimulation(plans, perturbations=<factory>, account_id='acct-sim', account_role='CustomerSubledger', parent_role='CustomerLedger', opening_balance=0.0, prefix='spec_example', rng=<factory>, emit_legs=True)[source]

Bases: object

A leaf internal account stepped forward day by day.

Authoring shape mirrors the AP.2 spike: plans declare per-day flows, perturbations declare the AP.2 perturbation knobs. The fold is PURE (_fold() is side-effect free); run(conn) writes the rows; violation_trajectory(inv, conn) carries the violation set as state day by day.

Per the AS.1 RNG convention: every concrete generator that composes AccountSimulation passes an rng (seeded via scenario_rng); AS.3 itself doesn’t randomize anything (the legs are author-declared), but the field carries forward for compose- time choices in AT’s anomaly/money_trail.

Parameters:

plans (list[DayPlan])
perturbations (list[Perturbation])
account_id (str)
account_role (str)
parent_role (str)
opening_balance (float)
prefix (str)
rng (Random)
emit_legs (bool)

account_id: str = 'acct-sim'

account_role: str = 'CustomerSubledger'

emit(conn, *, scenario_id=None)[source]

Write the full fold to the connection in one pass.

AV.5: scenario_id kwarg tags each emitted row’s metadata column for ScenarioContext cleanup attribution. None (the default) preserves the pre-AV.5 untagged emit shape.

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

emit_legs: bool = True: AS.4 — when False, the fold still computes per-day stored via Σ legs, but does NOT insert leg rows into _transactions. Right for parent-style ledger accounts that have no direct postings of their own — pure aggregators of child balances. Default True keeps every existing leaf-style use site unchanged. (Parents with mixed direct postings + child rollups also work; AO.L fixed _computed_ledger_balance to sum direct postings cumulatively across days, so the matview no longer relies on direct_totals=0 for clean drift.)

opening_balance: float = 0.0

parent_role: str = 'CustomerLedger'

perturbations: list[Perturbation]

plans: list[DayPlan]

prefix: str = 'spec_example'

rng: Random

violation_trajectory(invariant, conn)[source]

Run the fold day by day, refresh + detect after each day, return the per-day violation set. The carried-state shape from AP.2: each snapshot is the active violations as the institution reaches that day. The delta between consecutive snapshots IS each step’s effect (opened / closed / inert).

Return type:

list[set[Violation]]

Parameters:

invariant (Invariant)
conn (SyncConnection)

class recon_gen.common.spine.AnomalyGenerator(sender_account_id, sender_account_role, sender_account_parent_role, recipient_account_id, recipient_account_role, recipient_account_parent_role, anchor_day, spike_magnitude, baseline_pair_count, baseline_amount, prefix='spec_example')[source]

Bases: object

Plant a baseline distribution + a spike between sender ↔ recipient.

Emits baseline_pair_count extra pairs of background accounts with small uniform amounts on the anchor day (populates the matview’s pop_stddev) plus ONE spike pair (sender → recipient) with spike_magnitude (sits far above baseline → high z-score → fires).

Per AP.3 finding #2 (statistical invariants are multi-row by nature): the generator’s emit() writes ALL the rows in one call — the Protocol stays minimal; the per-row-iterator shape isn’t pushed onto the Generator contract.

AT.3 refactor: pairs are now emitted as Transfer`s through a transfers-only `LedgerSimulation. Single-edge property preserved (no AccountSimulation folds → no balance rows → no drift trip). Each baseline pair = one Posted 2-leg balanced Transfer; the spike is the same shape with spike_magnitude. Shape is identical to MoneyTrailGenerator’s — both consume the AT.3 primitive.

Parameters:

sender_account_id (str)
sender_account_role (str)
sender_account_parent_role (str | None)
recipient_account_id (str)
recipient_account_role (str)
recipient_account_parent_role (str)
anchor_day (date)
spike_magnitude (float)
baseline_pair_count (int)
baseline_amount (float)
prefix (str)

anchor_day: date

baseline_amount: float

baseline_pair_count: int

property claimed_accounts: frozenset[str]: The 2 + 2*baseline_pair_count account_ids this plant touches: the spike pair + every baseline pair’s sender/recipient. Used by AV.5 ScenarioContext.compose to catch cross-generator collisions at the wiring site.

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation

prefix: str = 'spec_example'

recipient_account_id: str

recipient_account_parent_role: str

recipient_account_role: str

sender_account_id: str

sender_account_parent_role: str | None

sender_account_role: str

spike_magnitude: float

class recon_gen.common.spine.AnomalyInvariant(prefix='spec_example')[source]

Bases: object

Pair-rolling-anomaly detector. Reads <prefix>_inv_pair_rolling_anomalies and projects EVERY row as a Violation — every (pair, window_end) the matview computed, across every z_bucket (including ‘0-1 sigma’ background).

Per AP.3 finding #3, the σ threshold belongs on the View, not the detector. AT.2 promoted AnomalyView (anomaly_view.py) that slices over the detected violation set on sigma_threshold. The detector here is now bucket-agnostic — AnomalyView(3.0).slice(…) reproduces AT.1’s behaviour exactly; other thresholds (2.0 for deep-dive triage, etc.) work over the same detect() result with no re-query.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'inv_pair_rolling_anomalies'

prefix: str = 'spec_example'

scenario_for(sender_role, recipient_role, *, spike_magnitude=100000.0, baseline_pair_count=100, baseline_amount=100.0, anchor_day=datetime.date(2030, 1, 1), instance=None, sender_account_id=None, recipient_account_id=None)[source]

Resolve sender + recipient roles; return a generator that plants baseline_pair_count baseline pairs + 1 spike between sender + recipient.

See AT.0 spike’s docstring for the full statistical-coverage argument. The defaults (100 baseline / 100_000 spike) give a clear ~10σ separation; tweak for tests that explore the threshold boundary (set spike=baseline to defuse).

Raises ValueError if either role is missing from the shape’s internal accounts (sender) or leaf internal accounts (recipient — the matview’s recipient filter requires account_parent_role IS NOT NULL).

AY.4.c — sender_account_id / recipient_account_id override the default synthetic IDs. The plant adapter (AY.4.c.3) threads OLD AnomalyPlant account_ids through these kwargs so N anomaly plants on the same (sender_role, recipient_role) pair produce N distinct generators (the default f”acct-anomaly-{sender,recipient}-{role}” derivations would collide). Existing test callers can pass nothing → preserves the synthetic defaults byte-stable.

Return type:

AnomalyGenerator

Parameters:

sender_role (str)
recipient_role (str)
spike_magnitude (float)
baseline_pair_count (int)
baseline_amount (float)
anchor_day (date)
instance (L2Instance | None)
sender_account_id (str | None)
recipient_account_id (str | None)

class recon_gen.common.spine.AnomalyView(sigma_threshold=3.0)[source]

Bases: object

Analyst-facing slice over the anomaly detector’s full output.

Holds the σ threshold for “include this bucket in the violation set.” A violation with bucket B is included iff BUCKET_LOWER_BOUNDS[B] >= sigma_threshold — i.e. the bucket’s lower edge is at-or-above the threshold. Defaults to 3.0 to match AT.1’s baked-in cutoff (analyst convention: “anomaly” starts at 3σ).

The View is pure (no IO; deterministic on its inputs); the detector still does the SQL read. slice(violations) is the only behaviour — other Views (depth-threshold for money_trail, etc.) will mirror this shape: pure projection over the detector’s output set.

Parameters:: sigma_threshold (float)

sigma_threshold: float = 3.0

slice(violations)[source]

Return the subset of violations whose z_bucket’s lower bound is ≥ sigma_threshold. Violations with no z_bucket key (defensive — non-anomaly invariants would be passed by mistake) are dropped silently; the caller’s job is to pass anomaly violations only.

Bucket strings not in BUCKET_LOWER_BOUNDS raise KeyError — that’s a matview-shape drift signal, not a normal runtime case, so it should fail loud rather than silently drop.

Return type:: set[Violation]
Parameters:: violations (set[Violation])

class recon_gen.common.spine.AuditFixture(invariant, identity)[source]

Bases: Violation

An audit-PDF input row marker.

AY.2.b’s 2 audit-fixture generators (SupersessionGenerator, FailedTransactionGenerator) emit these. Supersession + FailedTransaction rows surface only in the audit PDF (no matview, no coverage detector); the spine carries them for substrate uniformity (claimed_accounts collision check, AV.5 metadata tagging, ScenarioContext cleanup attribution).

Parameters:

invariant (str)
identity (frozenset[tuple[str, object]])

classmethod of(invariant, **identity)[source]

Backward-compat alias — returns a RuleViolation (a Violation subtype).

Pre-AY.2.a, every spine generator + detector built Violations via this constructor; AY.2.a layered in subtypes but kept the alias so the migration is incremental. New code should prefer the subtype-explicit constructors (RuleViolation.of(...) etc.) for type clarity.

Return type is the abstract Violation for liskov / covariant override compatibility — subclasses’ .of() can return their own concrete subtype without tripping pyright. At runtime this method returns a RuleViolation (the AS-era default).

Return type:

AuditFixture

Parameters:

invariant (str)
identity (object)

class recon_gen.common.spine.ChainCompletionGenerator(parent_transfer_id, parent_name, account_id, account_role, account_scope, account_parent_role, anchor_day, instance, prefix='spec_example')[source]

Bases: object

Emit one synthetic child leg per chain whose parent matches parent_name, with transfer_parent_id = parent_transfer_id so the chain matview sees a matched child.

Picks the FIRST non-fan_in child of each matching chain (deterministic — matches the OLD _baseline_xor_child_pick’s first-pick behavior for the AY.4.g minimum-viable scope; the fan_in / multi-pick variants land if AY.5 surfaces matview rows that need them).

Account fields denormalize onto the child leg from the parent plant’s account context — the matview keys on (transfer_parent_id, child_name), NOT account columns, so any account is fine. The adapter threads the parent plant’s account triple through.

intended returns a CoverageObservation: “I planted a synthetic child leg satisfying chain X.” No matching Invariant (the completion is a non-violating shape; coverage detector deferred).

No-op when parent_name parents no chain (the common case — most parent plants don’t sit on a chain-parent rail). The emit returns silently in that case.

Parameters:

parent_transfer_id (str)
parent_name (str)
account_id (str)
account_role (str)
account_scope (str)
account_parent_role (str | None)
anchor_day (date)
instance (L2Instance)
prefix (str)

account_id: str

account_parent_role: str | None

account_role: str

account_scope: str

anchor_day: date

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

instance: L2Instance

property intended: CoverageObservation

parent_name: str

parent_transfer_id: str

prefix: str = 'spec_example'

class recon_gen.common.spine.ChainParentDisagreementGenerator(child_template_name, anchor_day, parent_a_transfer_id='tr-cpd-parent-a', parent_b_transfer_id='tr-cpd-parent-b', rail_name='_spine_plant', prefix='spec_example', account_id_override=None)[source]

Bases: object

Emit 2 Posted transaction legs sharing one transfer_id + template_name but assigning different transfer_parent_id values — surfaces in the matview’s COUNT(DISTINCT transfer_parent_id) > 1 branch.

Account fields are synthetic + deterministic (the matview’s GROUP BY ignores them). The transfer_id + template_name combination IS the violation’s identity; both are derived from child_template_name so two generators built for the same template would collide at compose time — caught by AV.5’s claimed_accounts pairwise-disjoint check.

Single-edge: transfers-only emit → no balance rows → no drift trip (matches the AT.3 anomaly / money_trail shape).

AY.4.c.2 — account_id_override allows the plant adapter (AY.4.c.3) to thread OLD plant account_ids through, preventing PK collisions when N plants of the same shape compose.

Parameters:

child_template_name (str)
anchor_day (date)
parent_a_transfer_id (str)
parent_b_transfer_id (str)
rail_name (str)
prefix (str)
account_id_override (str | None)

property account_id: str: Single synthetic account_id the generator’s 2 legs land on. Matview filters don’t depend on account columns; the account is here just to satisfy NOT NULL constraints + AV.5’s claimed_accounts contract. account_id_override wins when set (AY.4.c.2 — plant-adapter PK-collision avoidance).

account_id_override: str | None = None

anchor_day: date

child_template_name: str

property claimed_accounts: frozenset[str]: Single synthetic account — AV.5 contract. Two generators targeting the same child_template_name collide here (same account_id derivation → same string in the claimed set).

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation: The natural-key tuple the matview surfaces post-plant.

parent_a_transfer_id: str = 'tr-cpd-parent-a'

parent_b_transfer_id: str = 'tr-cpd-parent-b'

prefix: str = 'spec_example'

rail_name: str = '_spine_plant'

property transfer_id: str: Deterministic child transfer_id — used as the matview’s natural-key tuple value + the row’s id discriminator.

class recon_gen.common.spine.ChainParentDisagreementInvariant(prefix='spec_example')[source]

Bases: object

Detector for the AB.2.3 matview.

Identity tuple: (transfer_id, child_template_name). The matview’s other columns (distinct_parent_count, parent_transfer_id_min, parent_transfer_id_max, business_day) are diagnostic — they help an analyst eyeball which parents disagreed but they’re not part of the violation’s identity.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'chain_parent_disagreement'

prefix: str = 'spec_example'

scenario_for(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Pick a chain whose singleton child is a TransferTemplate from the L2; return a generator that plants ONE child Transfer with 2 legs carrying disagreeing transfer_parent_id values.

Raises ValueError if the L2 has no chain whose singleton child resolves to a TransferTemplate (the AB.2.6 picker’s input requirement).

Return type:

ChainParentDisagreementGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

class recon_gen.common.spine.ClaimedAccountsGenerator(*args, **kwargs)[source]

Bases: Protocol

Extension of ViolationGenerator: declares the account_ids the plant will touch + accepts a scenario_id kwarg on emit.

The Protocol stays minimal — concrete generators add this property derived from their construction kwargs. emit’s scenario_id is keyword-only and defaults to None (preserves untagged behavior for existing call sites).

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: Violation | None

class recon_gen.common.spine.CoverageObservation(invariant, identity)[source]

Bases: Violation

A seed-color presence claim.

AY.2.b’s 5 coverage generators (TwoTemplateChainGenerator, TransferTemplateGenerator, RailFiringGenerator, InvFanoutGenerator, FanInChainGenerator(healthy)) emit these. Their intended carries the row the matching coverage detector (when one exists) reads back to confirm the seed met the documented coverage shape. ABSENCE is the regression — a CoverageInvariant.detect() returning empty means the seed didn’t emit what the docs claim.

Parameters:

invariant (str)
identity (frozenset[tuple[str, object]])

classmethod of(invariant, **identity)[source]

Backward-compat alias — returns a RuleViolation (a Violation subtype).

Pre-AY.2.a, every spine generator + detector built Violations via this constructor; AY.2.a layered in subtypes but kept the alias so the migration is incremental. New code should prefer the subtype-explicit constructors (RuleViolation.of(...) etc.) for type clarity.

Return type is the abstract Violation for liskov / covariant override compatibility — subclasses’ .of() can return their own concrete subtype without tripping pyright. At runtime this method returns a RuleViolation (the AS-era default).

Return type:

CoverageObservation

Parameters:

invariant (str)
identity (object)

class recon_gen.common.spine.DayEmission(day, legs, stored)[source]

Bases: object

The materialized result of one folded step: the legs to write and the stored balance that IS the running State'. run iterates these; violation_trajectory writes one at a time + refreshes + detects between.

Parameters:

day (date)
legs (tuple[tuple[str, float], ...])
stored (float)

day: date

legs: tuple[tuple[str, float], ...]

stored: float

class recon_gen.common.spine.DayPlan(day, legs)[source]

Bases: object

One day’s intended activity — signed leg amounts. A clean fold sets State'.balance = State.balance + Σ legs and stores it.

Parameters:

day (date)
legs (tuple[float, ...])

day: date

legs: tuple[float, ...]

class recon_gen.common.spine.DriftGenerator(child_account_id, child_role, parent_role, parent_account_id, parent_account_role, anchor_day, magnitude, rng=<factory>, leg_amount=100.0, prefix='spec_example')[source]

Bases: object

Emit a child account whose stored money drifts from its leg-total by magnitude, plus a parent account whose stored money equals the CLEAN leg-total. Result: child drifts (stored−computed = magnitude) AND parent drifts (parent.stored − Σ child.money = −magnitude).

The pre-AS.3 simple shape: one day, one account pair, one leg. AS.3 promotes this to a stateful day-by-day fold; AS.4 generalizes to cross-account vector state.

Parameters:

child_account_id (str)
child_role (str)
parent_role (str)
parent_account_id (str | None)
parent_account_role (str | None)
anchor_day (date)
magnitude (float)
rng (Random)
leg_amount (float)
prefix (str)

property also_trips_ledger_drift: RuleViolation | None

when this generator’s child-drift propagates up to the parent’s _ledger_drift. None when no parent account is present in the shape (the L2 instance has no account with the child’s parent_role).

Type:: The secondary edge

anchor_day: date

child_account_id: str

child_role: str

property claimed_accounts: frozenset[str]: The child account_id this plant drifts + the parent account_id when the shape has one (ledger_drift edge fires on the parent). AV.5.

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation

leg_amount: float = 100.0: Clean leg amount; the child’s stored money is this + magnitude.

magnitude: float

parent_account_id: str | None

parent_account_role: str | None

parent_role: str

prefix: str = 'spec_example'

rng: Random

class recon_gen.common.spine.DriftInvariant(prefix='spec_example')[source]

Bases: object

Sub-ledger drift detector. Persona-blind (no L2 join in the matview SQL), so scenario_for(role) resolves against any leaf internal account with that role.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'drift'

prefix: str = 'spec_example': Prefix of the deployed L2 instance’s matviews. Concrete invariants carry this so detect() can read the right matview; AS.1’s Protocol stayed minimal (no prefix field) since not every invariant variant needs one.

scenario_for(role, *, magnitude=5.0, seed=None, instance=None, child_account_id=None, parent_account_id=None)[source]

Resolve the role against the shape and return a generator that manufactures a drift breach on a leaf account of that role.

instance=None loads the bundled spec_example; AS.x callers thread the real instance.

AY.4.c — child_account_id / parent_account_id override the default synthetic IDs. The plant adapter (AY.4.c.3) threads OLD DriftPlant.account_id through these kwargs so N drift plants on the same role produce N distinct generators (the default f”acct-drift-child-{role}” derivation would collide). Existing test callers can pass nothing → preserves the synthetic defaults byte-stable.

Return type:

DriftGenerator

Parameters:

role (str)
magnitude (float)
seed (int | None)
instance (L2Instance | None)
child_account_id (str | None)
parent_account_id (str | None)

class recon_gen.common.spine.ExpectedEodBalanceGenerator(account_id, account_role, account_parent_role, anchor_day, expected, variance, prefix='spec_example')[source]

Bases: object

Emit a daily_balances row whose money is the configured expected ± variance, with expected_eod_balance = expected. NO transactions — the variance matview reads daily_balances directly.

variance=0.0 ⇒ money == expected ⇒ no variance row materializes. Non-violating shape per the AP.2 convention.

AU.0 finding: on a LEAF internal account (account_parent_role != None), this emission ALSO trips DriftInvariant, because drift’s matview filter parent_role IS NOT NULL AND stored ≠ Σ legs is satisfied (no transactions ⇒ Σ legs = 0; planted stored = expected + variance ≠ 0). Registry records the two-edge entry.

Parameters:

account_id (str)
account_role (str)
account_parent_role (str | None)
anchor_day (date)
expected (float)
variance (float)
prefix (str)

account_id: str

account_parent_role: str | None

account_role: str

property also_trips_drift: RuleViolation | None

drift fires on the same account/day when the planted account is a LEAF (account_parent_ role is set). Drift magnitude = stored − Σ legs = (expected + variance) − 0 = expected + variance.

Returns None when the planted account is NOT a leaf — drift’s matview filter excludes parent-role rows.

Type:: The empirical AU.0-style edge

anchor_day: date

property claimed_accounts: frozenset[str]: The single account_id this plant carries an EOD target for. AV.5.

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

expected: float

property intended: RuleViolation

prefix: str = 'spec_example'

variance: float

class recon_gen.common.spine.ExpectedEodBalanceInvariant(prefix='spec_example')[source]

Bases: object

Expected-EOD-balance detector. Persona-blind — the matview SQL filters only on the per-row expected_eod_balance column being set and not matching money. scenario_for(role) accepts ANY internal account with the requested role (no leaf/parent filter).

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'expected_eod_balance_breach'

prefix: str = 'spec_example'

scenario_for(role, *, expected=100.0, variance=5.0, instance=None, account_id=None)[source]

Resolve a role; return a generator that plants money = expected + variance with the per-row expected_eod_balance set, so the variance row materializes.

variance=0.0 is the non-violating shape (stored == expected ⇒ the matview row is filtered out). Same AP.2 convention as overdraft / drift.

Raises ValueError if the L2 has no internal account with the requested role.

AY.4.c — account_id overrides the default synthetic ID. The plant adapter (AY.4.c.3) threads OLD ExpectedEodBalancePlant.account_id through this kwarg so N plants on the same role produce N distinct generators (the default f”acct-eod-{role}” derivation would collide). Existing test callers can pass nothing → preserves the synthetic default byte-stable.

Return type:

ExpectedEodBalanceGenerator

Parameters:

role (str)
expected (float)
variance (float)
instance (L2Instance | None)
account_id (str | None)

class recon_gen.common.spine.FailedTransactionGenerator(account_id, account_role, account_scope, account_parent_role, rail_name, amount, anchor_day, prefix='spec_example')[source]

Bases: object

Plant ONE Debit leg with status='Failed'.

Single-leg, no counter-leg: a Failed transaction never settled, so the rail’s other side wasn’t created. The leg posts at a normal Debit shape with status='Failed' so the L2FT postings dataset’s CASE WHEN status IN ('Pending','Posted') THEN status ELSE 'Other' END collapses it to Other.

Account context (role / scope / parent_role) arrives as construction args; the AY.4 adapter resolves them from the OLD plant’s referenced template instance.

Parameters:

account_id (str)
account_role (str)
account_scope (str)
account_parent_role (str | None)
rail_name (str)
amount (float)
anchor_day (date)
prefix (str)

account_id: str

account_parent_role: str | None

account_role: str

account_scope: str

amount: float

anchor_day: date

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: AuditFixture: The Failed-status row’s identity — the dropdown-coverage claim the X.1.g e2e test reads back.

prefix: str = 'spec_example'

rail_name: str

property transaction_id: str: Deterministic id from the account_id — claimed_accounts + transaction_id naming both derive purely from construction fields so the AV.5 ScenarioContext collision check fires consistently.

class recon_gen.common.spine.FanInChainGenerator(chain_parent_name, child_template_name, expected_parent_count, parent_count, anchor_day, expected_kind, prefix='spec_example', account_id_override=None)[source]

Bases: object

Emit parent_count synthetic parent legs + 1 child Transfer whose legs each carry a contributing parent’s transfer_parent_id. The matview reads COUNT(DISTINCT parent_transfer_id) for the child and compares to the L2’s declared expected_parent_count to derive the disagreement_kind row (or no row, in the healthy case).

Account fields are synthetic — the matview doesn’t filter on them. The parent_count knob differentiates the 3 variants: healthy (== expected), missing (< expected), extra (> expected).

Single-edge: transfers-only → no balance rows → no drift trip.

AY.4.c.2 — account_id_override allows the plant adapter (AY.4.c.3) to thread OLD plant account_ids through, preventing PK collisions when N plants of the same shape compose.

Parameters:

chain_parent_name (str)
child_template_name (str)
expected_parent_count (int | None)
parent_count (int)
anchor_day (date)
expected_kind (str)
prefix (str)
account_id_override (str | None)

property account_id: str: Derivation keys off expected_kind + child_template_name (variant kind matters here — healthy/missing/orphan/extra each get a distinct account so a compose of multiple variants on the same chain doesn’t PK-collide). account_id_override wins when set (AY.4.c.2).

account_id_override: str | None = None

anchor_day: date

chain_parent_name: str

child_template_name: str

property child_transfer_id: str

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

expected_kind: str

expected_parent_count: int | None

property intended: Violation

The evidence this plant produces:

For ‘missing’ / ‘orphan’ / ‘extra’ variants — a RuleViolation (the matview surfaces the disagreement row).
For ‘healthy’ (parent_count == expected) — a CoverageObservation (the AY.2.b layering: the plant emits a chain firing that doesn’t trip the fan_in matview but DOES populate the chain’s parent_count seed coverage). Pre-AY.2.b this was None; AY.2.b promotes it to a typed CoverageObservation so the seed’s “I planted a healthy fan-in chain” claim has somewhere to land.

parent_count: int

prefix: str = 'spec_example'

class recon_gen.common.spine.FanInDisagreementInvariant(prefix='spec_example')[source]

Bases: object

Detector for the AB.4.7 matview.

Identity tuple: (child_transfer_id, disagreement_kind). Other columns (chain_parent_name, child_template_name, parent_count, expected_parent_count, business_day) are diagnostic.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'fan_in_disagreement'

prefix: str = 'spec_example'

scenario_for_extra_parent(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Extra-parent shape: parent_count = expected + 1. Only defined when expected is set on the L2 chain — the matview’s ‘extra’ branch gates on expected IS NOT NULL.

Raises ValueError if the picked chain has no expected_parent_count (the variable-batch case has no upper bound to exceed).

Return type:

FanInChainGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

scenario_for_healthy(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Healthy non-violating shape per AP.2: parent_count == expected (when set; defaults to 2 for variable-batch). Emit produces NO matview row — the AP.2 convention’s positive-control for the test suite.

Return type:

FanInChainGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

scenario_for_missing_parent(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Missing-parent shape: parent_count = expected - 1 (when expected is set) OR parent_count = 1 (orphan, when expected is unset on the L2 chain — variable-batch case where ≥2 is the implicit floor).

Disagreement_kind on the matview row: ‘missing’ when expected is set; ‘orphan’ when expected is unset and parent_count < 2.

Return type:

FanInChainGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

class recon_gen.common.spine.InvFanoutFactory(prefix='spec_example')[source]

Bases: object

Smart constructor namespace for InvFanoutGenerator.

No L2-instance resolution needed — the Investigation matviews don’t read rail/template declarations (they walk transactions by account_scope + account_parent_role + transfer_id). The factory just supplies sensible defaults + synthetic accounts.

Parameters:: prefix (str)

name: ClassVar[str] = 'inv_fanout'

prefix: str = 'spec_example'

scenario_for_fanout(*, sender_count=2, recipient_account_id='acct-inv-fanout-recipient', rail_name='_spine_plant', amount_per_transfer=100.0, anchor_day=datetime.date(2030, 1, 1))[source]

Build a generator that plants sender_count distinct senders all crediting one recipient on the anchor day.

Defaults are minimal-viable: 2 senders × $100 stays under the anomaly z-score threshold (the OLD plant’s boost_inv_ fanout_plants(5×) multiplies for anomaly density; the spine generator delegates that decision to the picker layer).

Return type:

InvFanoutGenerator

Parameters:

sender_count (int)
recipient_account_id (str)
rail_name (str)
amount_per_transfer (float)
anchor_day (date)

class recon_gen.common.spine.InvFanoutGenerator(recipient_account_id, sender_account_ids, rail_name, amount_per_transfer, anchor_day, prefix='spec_example', recipient_role='CustomerSubledger', recipient_parent_role='CustomerLedger', sender_role='ExternalCounterparty', sender_scope='external', sender_parent_role=None)[source]

Bases: object

Emit N two-leg transfers (one per sender) all crediting one leaf-internal recipient on the anchor day. Each transfer is a debit on the sender + credit on the recipient summing to zero, sharing one transfer_id, both legs stamped with metadata.sender_id / recipient_id for downstream investigation.

Recipient’s account fields (role, parent_role) MUST satisfy the inv_pair_rolling_anomalies matview filter (account_scope=’internal’ AND account_parent_role IS NOT NULL) — defaults below populate them; production picker threads real TemplateInstance.account_id / role tuples post-AY.4.

intended returns a CoverageObservation. Registered edge to MoneyTrailInvariant: each transfer is a depth-0 edge the inv_money_trail_edges recursive CTE surfaces.

Parameters:

recipient_account_id (str)
sender_account_ids (tuple[str, ...])
rail_name (str)
amount_per_transfer (float)
anchor_day (date)
prefix (str)
recipient_role (str)
recipient_parent_role (str)
sender_role (str)
sender_scope (str)
sender_parent_role (str | None)

amount_per_transfer: float

anchor_day: date

property claimed_accounts: frozenset[str]

recipient + every sender. Two fanouts to the same recipient on the same day would collide via the recipient account_id.

Type:: AV.5 contract

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: CoverageObservation

a fanout of sender_count senders to recipient_account_id landed on anchor_day. Identity carries the natural-key tuple a coverage detector would round-trip against if/when one lands.

Type:: Presence evidence

prefix: str = 'spec_example'

rail_name: str

recipient_account_id: str

recipient_parent_role: str = 'CustomerLedger'

recipient_role: str = 'CustomerSubledger'

sender_account_ids: tuple[str, ...]

sender_parent_role: str | None = None

sender_role: str = 'ExternalCounterparty'

sender_scope: str = 'external'

property transfer_ids: tuple[str, ...]: The len(sender_account_ids) transfer_ids this plant emits, in stable sender-sorted order (matches the OLD emitter’s deterministic ordering).

class recon_gen.common.spine.Invariant(*args, **kwargs)[source]

Bases: Protocol

A rule + detector. Concrete invariants implement this via:

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.
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]

class recon_gen.common.spine.LedgerDriftInvariant(prefix='spec_example')[source]

Bases: object

Parent-ledger drift detector. Fires when a parent’s stored money ≠ Σ(child stored) + Σ(direct legs). DriftGenerator’s child-drift causes Σ(child stored) to shift — so this fires on the parent too.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'ledger_drift'

prefix: str = 'spec_example'

class recon_gen.common.spine.LedgerSimulation(accounts=<factory>, transfers=<factory>, prefix='spec_example')[source]

Bases: object

A vector of `AccountSimulation`s sharing one connection, plus an optional list of cross-account `Transfer`s.

Each per-account fold is independent (scalar); the LEDGER’s cross-account behavior emerges from the matview SQL — e.g., ledger_drift’s Σ child.money reads every account’s _current_daily_balances row. So vector state here is “many scalar folds emitted side by side”; the cross-boundary invariants pick up the structural property from the data.

transfers (AT.3) is the cross-account FLOW dimension — transfer-shaped emissions (multi-leg, shared transfer_id, optional parent_transfer_id for chains). Anomaly’s pair plant is “transfers only, no accounts” (no balance rows → single-edge to anomaly). Money_trail’s recursive chain is “transfers with parent linkage”. Drift-style scenarios continue to use the account-fold shape; a scenario that wants both composes both fields.

Per the AS.1 RNG convention: each composed AccountSimulation carries its own seeded rng. LedgerSimulation doesn’t override.

Parameters:

accounts (list[AccountSimulation])
transfers (list[Transfer])
prefix (str)

accounts: list[AccountSimulation]

emit(conn, *, scenario_id=None)[source]

Write every account’s full fold AND every transfer’s legs. Commits to the caller — so a scenario can compose multiple LedgerSimulations against one connection and refresh once at the end (the AP.3 pattern).

AV.5: scenario_id kwarg threads through to the per-row metadata tag ({"scenario_id": "..."}) when set; None preserves the pre-AV.5 untagged emit (byte-identical).

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

prefix: str = 'spec_example': Prefix for the <prefix>_transactions table when emitting transfers. For account-only ledgers this is unused (each AccountSimulation carries its own prefix). For transfer-only ledgers (anomaly’s shape) this is the source of truth.

transfers: list[Transfer]

violation_trajectory(invariant, conn)[source]

Per-day violation snapshots, like AccountSimulation’s but emitting all accounts’ rows for day i BEFORE refresh+detect.

Assumes every AccountSimulation in the ledger has the SAME number of plans (one DayPlan per ledger-day). Heterogeneous timelines would need a different fold; AT.x extensions can add a sparser shape when there’s a use case.

Return type:

list[set[Violation]]

Parameters:

invariant (Invariant)
conn (SyncConnection)

class recon_gen.common.spine.LimitBreachGenerator(account_id, account_role, account_parent_role, rail_name, direction, cap, overshoot, anchor_day, prefix='spec_example')[source]

Bases: object

Emit a single Posted transaction whose ABS(amount_money) = cap + overshoot for the given (account, day, rail, direction). The matview’s GROUP BY collapses to this single row → SUM = cap + overshoot > cap → limit_breach fires.

Sign convention is locked by the transactions CHECK constraint — Debit ⇒ money ≤ 0; Credit ⇒ money ≥ 0. The matview’s SUM(ABS) makes both contribute positively.

Parameters:

account_id (str)
account_role (str)
account_parent_role (str)
rail_name (str)
direction (Literal['Outbound', 'Inbound'])
cap (float)
overshoot (float)
anchor_day (date)
prefix (str)

account_id: str

account_parent_role: str

account_role: str

anchor_day: date

cap: float

property claimed_accounts: frozenset[str]: The single account_id this plant breaches a cap on. AV.5.

direction: Literal['Outbound', 'Inbound']

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation

overshoot: float

prefix: str = 'spec_example'

rail_name: str

class recon_gen.common.spine.LimitBreachInvariant(prefix='spec_example')[source]

Bases: object

Per-rail per-direction flow-cap detector. The matview gates on cap IS NOT NULL (rail+parent_role+direction has a LimitSchedule) AND SUM(ABS(amount_money)) > cap. Identity is (account_id, business_day, rail_name, direction) — analyst-facing diff readability.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'limit_breach'

prefix: str = 'spec_example'

scenario_for(parent_role, rail_name, *, direction='Outbound', overshoot=100.0, instance=None, account_id=None)[source]

Resolve (parent_role, rail_name, direction) against the L2’s LimitSchedule; return a generator that plants ONE Posted transaction on a child account whose account_parent_role = parent_role, with amount_money sized to overshoot the cap.

overshoot=0.0 ⇒ amount == cap ⇒ matview’s strict > filter excludes ⇒ no fire (AP.2 non-violating convention adapted to Money-unit knob). Positive fires.

Return type:

LimitBreachGenerator

Parameters:

parent_role (str)
rail_name (str)
direction (Literal['Outbound', 'Inbound'])
overshoot (float)
instance (L2Instance | None)
account_id (str | None)

Raises ValueError if: - The L2 has no LimitSchedule matching `(parent_role, rail_name,

direction)`

The L2 has no child account with account_parent_role = parent_role (the matview filters account_parent_role IS NOT NULL; without a matching child the plant is inert)

AY.4.c — account_id overrides the default synthetic ID. The plant adapter (AY.4.c.3) threads OLD LimitBreachPlant.account_id through this kwarg so N plants on the same (parent_role, rail, direction) triple produce N distinct generators (the default f”acct-limit-breach-{rail_name}-{direction}” derivation would collide). Existing test callers can pass nothing → preserves the synthetic default byte-stable.

Note: LimitBreachGenerator carries only one account_id field (the breaching account); there is no counter_account_id — the matview groups solely on (account_id, business_day, rail_name, direction).

class recon_gen.common.spine.MoneyTrailGenerator(hop_account_role, hop_account_parent_role, chain_length, amount, anchor_day, prefix='spec_example', chain_id_prefix='money-trail')[source]

Bases: object

Plant a parent-linked chain of chain_length Posted transfers.

Each transfer is a 2-leg balanced send: account[i] → account[i+1]. Transfer[i+1].parent_transfer_id = Transfer[i].transfer_id (the chain linkage the matview walks recursively). All transfers post on consecutive days from anchor_day (anchor_day, +1d, +2d, …) so the matview’s posted_at ordering matches chain depth.

Per AP.3 finding #2 generalized: graph invariants are multi-row by nature — N transfers + N+1 accounts per chain. Single-row plant would yield depth=0 only (not a “chain” in any analytically meaningful sense). The generator’s emit() writes ALL the rows in one call through a transfers-only LedgerSimulation.

Single-edge property: no AccountSimulation folds → no balance rows → no drift trip from the chain. Matches AnomalyGenerator’s AT.0-finding.

Parameters:

hop_account_role (str)
hop_account_parent_role (str)
chain_length (int)
amount (float)
anchor_day (date)
prefix (str)
chain_id_prefix (str)

amount: float

anchor_day: date

chain_id_prefix: str = 'money-trail': AY.4.c — disambiguator threaded into _account_id / _transfer_id so N money-trail plants on the same hop_role produce N distinct chains. Defaults to the legacy “money-trail” value → existing test callers stay byte-stable.

chain_length: int

property claimed_accounts: frozenset[str]: The chain_length+1 hop account_ids the chain walks through. account[0] is the chain’s root sender; account[N] is the leaf recipient. Used by AV.5 ScenarioContext.compose to catch cross-generator collisions at the wiring site.

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

hop_account_parent_role: str

hop_account_role: str

property intended: RuleViolation

The deepest edge of the chain — the “story” of the trail. For chain_length=3, depth=2 (grandchild) is the most-removed- from-root edge, the analyst-meaningful endpoint.

Violation identity also includes transfer_id + root_transfer_id, both deterministic from this generator’s account/chain configuration.

prefix: str = 'spec_example'

class recon_gen.common.spine.MoneyTrailInvariant(prefix='spec_example')[source]

Bases: object

Money-trail edge detector. Reads <prefix>_inv_money_trail_edges and projects every edge as a Violation with identity (root_transfer_id, transfer_id, depth).

The detector is bucket-agnostic — every edge is returned; the MoneyTrailView (this module) slices on min_depth for analyst- facing “suspicious chain” thresholds. Mirrors AT.2’s AnomalyInvariant ⋈ AnomalyView shape.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'inv_money_trail_edges'

prefix: str = 'spec_example'

scenario_for(hop_role, *, chain_length=3, amount=100.0, anchor_day=datetime.date(2030, 1, 1), instance=None, chain_id_prefix=None)[source]

Resolve the chain’s account role + return a generator that plants a chain_length-deep parent-linked chain. Every account in the chain uses hop_role (the role of the leaf-eligible internal accounts the matview walks).

chain_length = number of transfers in the chain (depth runs 0..N-1). chain_length=1 = single transfer (no parent linkage, depth=0 only). chain_length=3 = root → child → grandchild (depths 0/1/2).

Each transfer’s recipient leg becomes the next transfer’s sender — money walks through a chain of chain_length + 1 distinct accounts. The matview surfaces each transfer as one edge (one source-leg × one target-leg per transfer).

AY.4.c — chain_id_prefix overrides the default synthetic ID prefix. Unlike the single-account spine factories, money_trail plants chain_length + 1 account_ids + chain_length transfer_ids; the natural disambiguator is a chain-wide prefix feeding both _account_id(i) and _transfer_id(i). The plant adapter (AY.4.c.3) threads OLD money-trail chain identifiers through this kwarg so N plants on the same hop_role produce N distinct generators (the default “acct-money-trail-hop” / “xfer-money-trail” derivations would collide). Existing test callers can pass nothing → preserves the synthetic defaults byte-stable.

Return type:

MoneyTrailGenerator

Parameters:

hop_role (str)
chain_length (int)
amount (float)
anchor_day (date)
instance (L2Instance | None)
chain_id_prefix (str | None)

class recon_gen.common.spine.MoneyTrailView(min_depth=0)[source]

Bases: object

Analyst-facing slice over the money-trail detector’s full output.

Holds the min_depth knob for “include edges whose chain depth is at-or-above this value”. min_depth=0 (default) includes every edge (matches the detector’s full return). min_depth=1 drops the root edge of every chain; min_depth=2 keeps only grandchild- and-deeper edges (the “this is a chain, not a one-off” view).

Pure (no IO; deterministic on its inputs); the detector still does the SQL read. Mirrors AnomalyView — the View pattern is the same for every L2 invariant.

Parameters:: min_depth (int)

min_depth: int = 0

slice(violations)[source]

Return the subset of violations whose depth is ≥ min_depth. Violations missing the depth key are dropped silently (defensive against cross-invariant mix).

Return type:: set[Violation]
Parameters:: violations (set[Violation])

class recon_gen.common.spine.MultiXorMissedGenerator(chain_parent_name, anchor_day, instance=None, prefix='spec_example', account_id_override=None)[source]

Bases: object

Plant a parent Transfer with NO declared XOR-sibling children firing. Matview’s child_count = 0 → ‘missed’ row.

Emits ONE leg row for the parent: rail_name stamps the chain.parent (or the template’s first leg_rail when the parent is a Template); template_name stamps the parent template name when applicable.

AY.4.c.2 — account_id_override allows the plant adapter (AY.4.c.3) to thread OLD plant account_ids through, preventing PK collisions when N plants of the same shape compose.

Parameters:

chain_parent_name (str)
anchor_day (date)
instance (L2Instance | None)
prefix (str)
account_id_override (str | None)

property account_id: str: account_id_override wins when set (AY.4.c.2).

account_id_override: str | None = None

anchor_day: date

chain_parent_name: str

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

instance: L2Instance | None = None

property intended: RuleViolation

property parent_transfer_id: str

prefix: str = 'spec_example'

class recon_gen.common.spine.MultiXorOverlapGenerator(chain_parent_name, variant_a_child_name, variant_b_child_name, anchor_day, instance=None, prefix='spec_example', account_id_override=None)[source]

Bases: object

Plant a parent Transfer + TWO child firings (both XOR-siblings; both linked to the parent via transfer_parent_id). Matview’s child_count = 2 → ‘overlap’ row.

Emits 3 leg rows: 1 parent + 2 children. Each child row’s rail_name/template_name matches the chain’s declared XOR-sibling child name (rail vs template kind resolved at emit time).

AY.4.c.2 — account_id_override allows the plant adapter (AY.4.c.3) to thread OLD plant account_ids through, preventing PK collisions when N plants of the same shape compose.

Parameters:

chain_parent_name (str)
variant_a_child_name (str)
variant_b_child_name (str)
anchor_day (date)
instance (L2Instance | None)
prefix (str)
account_id_override (str | None)

property account_id: str: account_id_override wins when set (AY.4.c.2).

account_id_override: str | None = None

anchor_day: date

chain_parent_name: str

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

instance: L2Instance | None = None

property intended: RuleViolation

property parent_transfer_id: str

prefix: str = 'spec_example'

variant_a_child_name: str

variant_b_child_name: str

class recon_gen.common.spine.MultiXorViolationInvariant(prefix='spec_example')[source]

Bases: object

Detector for the AB.6.5 matview.

Identity tuple: (parent_transfer_id, disagreement_kind). The matview’s other columns (parent_rail_or_template_name, child_count, fired_children, business_day) are diagnostic.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'multi_xor_violation'

prefix: str = 'spec_example'

scenario_for_missed(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Pick a chain with ≥2 non-fan_in children + return a generator that plants ONE parent firing with NO child firings — matview reads count=0, surfaces ‘missed’ row.

Return type:

MultiXorMissedGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

scenario_for_overlap(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Pick a chain with ≥2 non-fan_in children + return a generator that plants ONE parent firing + TWO child firings (both XOR-siblings; both linked to the parent via transfer_parent_id) — matview reads count=2, surfaces ‘overlap’ row.

Return type:

MultiXorOverlapGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

class recon_gen.common.spine.OverdraftGenerator(account_id, account_role, account_parent_role, anchor_day, magnitude, prefix='spec_example')[source]

Bases: object

Emit a daily_balances row whose money is below zero by magnitude. NO transactions — overdraft’s matview reads current_daily_balances directly; only the balance row is needed.

Per the AP.2 convention: magnitude=0.0 means the perturbation is OFF; the emitted row has money=0, which is NOT < 0, so overdraft does NOT fire. The non-violating shape is the same generator with the knob off.

AU.0 finding: on a LEAF internal account (account_parent_role != None), this emission ALSO trips DriftInvariant because drift’s matview filter parent_role IS NOT NULL AND stored ≠ Σ legs is satisfied (no transactions emitted ⇒ Σ legs = 0 ≠ −magnitude). The registry records the two-edge entry.

Parameters:

account_id (str)
account_role (str)
account_parent_role (str | None)
anchor_day (date)
magnitude (float)
prefix (str)

account_id: str

account_parent_role: str | None

account_role: str

property also_trips_drift: RuleViolation | None

drift fires on the same account/day when the planted account is a LEAF (account_parent_role is set). Returns None when the planted account is NOT a leaf (drift’s parent_role IS NOT NULL filter excludes it).

Magnitude sign: drift = stored − Σ legs = −magnitude − 0 = −magnitude.

Type:: The empirical AU.0 edge

anchor_day: date

property claimed_accounts: frozenset[str]: The single account_id this plant overdrafts. AV.5.

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation

magnitude: float

prefix: str = 'spec_example'

class recon_gen.common.spine.OverdraftInvariant(prefix='spec_example')[source]

Bases: object

Non-negative-stored-balance detector. Persona-blind — the matview SQL is WHERE money < 0 on every internal account, no role join. scenario_for(role) filters the L2 by role only; ANY scope=internal account qualifies (no parent_role requirement that drift carries).

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'overdraft'

prefix: str = 'spec_example': Prefix of the deployed L2 instance’s matviews. Same default + per-call override pattern drift uses.

scenario_for(role, *, magnitude=5.0, instance=None, account_id=None)[source]

Resolve a role against the shape; return a generator that manufactures a stored-balance overdraft on the first internal account with that role.

magnitude is caller-facing (“how far below zero the planted stored is” — positive). magnitude=0.0 plants stored=0 which is NOT < 0, so overdraft does NOT fire — AP.2’s non-violating convention promoted to overdraft.

Raises ValueError if the L2 has no internal account with the requested role. Smart-constructor discipline matching drift’s: the invariant owns shape resolution, fails loud at the request site, never silently emits inert rows.

instance=None loads the bundled spec_example — production callers (deploy-time, e2e fixtures) thread the real L2.

AY.4.c — account_id overrides the default synthetic ID. The plant adapter (AY.4.c.3) threads OLD OverdraftPlant.account_id through this kwarg so N overdraft plants on the same role produce N distinct generators (the default f”acct-overdraft-{role}” derivation would collide). Existing test callers can pass nothing → preserves the synthetic default byte-stable.

Return type:

OverdraftGenerator

Parameters:

role (str)
magnitude (float)
instance (L2Instance | None)
account_id (str | None)

class recon_gen.common.spine.Perturbation(kind='none', day_index=0, amount=0.0, correct_day_index=None)[source]

Bases: object

How a single day’s step deviates from the clean fold. The knob AP.2 surfaced — one shape, three kinds + a correction.

"none": clean — never emits anything on this day (no-op).
"state_blip": corrupt ONLY the stored snapshot on day_index by amount (the running balance stays clean → drift is LOCAL to that day).
"unrecorded_leg": emit an extra leg on day_index that is NOT folded into stored (flow/state DISAGREEMENT → drift PROPAGATES forward; computed is cumulative, stored stayed on the clean fold). Optional correct_day_index books the leg into stored on a later day (the AN.1 supersession shape — closes the forward propagation; the historical breach remains).
"recorded_leg": emit an EXTRA real leg that DOES fold into state — a different-but-consistent history (conforming).

Parameters:

kind (Literal['none', 'state_blip', 'unrecorded_leg', 'recorded_leg'])
day_index (int)
amount (float)
correct_day_index (int | None)

amount: float = 0.0

correct_day_index: int | None = None

day_index: int = 0

kind: Literal['none', 'state_blip', 'unrecorded_leg', 'recorded_leg'] = 'none'

class recon_gen.common.spine.RailFiringFactory(prefix='spec_example')[source]

Bases: object

Smart constructor namespace for RailFiringGenerator.

Mirrors the Invariant.scenario_for(…) pattern from the rest of the spine for surface parity. Picks a rail by name from the L2 instance + resolves its kind, then builds a generator with the right field shape.

Parameters:: prefix (str)

name: ClassVar[str] = 'rail_firing'

prefix: str = 'spec_example'

scenario_for_rail(rail_name, *, account_id_a=None, account_id_b=None, amount=100.0, firing_seq=1, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Resolve rail_name to a Rail + build a generator that plants one firing.

Account fields default to synthetic per-rail strings when not supplied — matches the AY.0 minimal-viable shape. The picker layer (post-AY.4) threads real TemplateInstance / Account identifiers when the firing needs to land on a materialized account.

Raises ValueError if the rail isn’t declared on the L2 OR if account_id_b is None for a TwoLegRail (second leg required for two-leg rails).

Return type:

RailFiringGenerator

Parameters:

rail_name (str)
account_id_a (str | None)
account_id_b (str | None)
amount (float)
firing_seq (int)
anchor_day (date)
instance (L2Instance | None)

class recon_gen.common.spine.RailFiringGenerator(rail_name, is_two_leg, account_id_a, account_id_b, amount, single_leg_direction, firing_seq, anchor_day, prefix='spec_example', metadata_extras=())[source]

Bases: object

Emit one Posted firing of an L2-declared Rail.

TwoLegRail: 2 legs (debit on account_id_a for -amount + credit on account_id_b for +amount) sharing one transfer_id; net = 0.

SingleLegRail: 1 leg on account_id_a in the resolved single_leg_direction; net = ±amount (the L1 SQL surfaces this as ‘Imbalanced’ against the rail’s expected_net = 0 — accurate representation of a bare single-leg cycle without its sibling legs in the broad-mode picker context).

intended returns a CoverageObservation keyed on (rail_name, transfer_id, firing_seq). No matching Invariant (rail firings aren’t violations).

Account fields are caller-supplied (the factory provides deterministic synthetic defaults). The picker layer (post-AY.4) threads real TemplateInstance.account_id / Account.id values when the firing needs to land on a materialized account.

Parameters:

rail_name (str)
is_two_leg (bool)
account_id_a (str)
account_id_b (str | None)
amount (float)
single_leg_direction (Literal['Debit', 'Credit'])
firing_seq (int)
anchor_day (date)
prefix (str)
metadata_extras (tuple[tuple[str, str], ...])

account_id_a: str

account_id_b: str | None

amount: float

anchor_day: date

property claimed_accounts: frozenset[str]

union of the rail’s leg account_ids. Two firings of the same rail with different firing_seq target the same account_id_a → ScenarioContext rejects parallel firings of the same rail unless the picker varies accounts.

Type:: AV.5 contract

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

firing_seq: int

property intended: CoverageObservation

rail X fired once on the anchor day. Identity carries the natural-key tuple a coverage detector would round-trip against if/when one lands.

Type:: Presence evidence

is_two_leg: bool

metadata_extras: tuple[tuple[str, str], ...] = ()

prefix: str = 'spec_example'

rail_name: str

single_leg_direction: Literal['Debit', 'Credit']

property transfer_id: str

class recon_gen.common.spine.RuleViolation(invariant, identity)[source]

Bases: Violation

A matview-detected rule break.

The post-AS shape — every L1 accounting / L2-shape integrity / L2 investigation invariant promoted pre-AY emits these. The semantic_lock dict’s values are frozenset[RuleViolation] for every spine Invariant shipped to date.

Parameters:

invariant (str)
identity (frozenset[tuple[str, object]])

classmethod of(invariant, **identity)[source]

Build a typed RuleViolation. The blessed constructor for rule-violation evidence.

Return type:

RuleViolation

Parameters:

invariant (str)
identity (object)

class recon_gen.common.spine.ScenarioContext(scenario_id, prefix='spec_example', dialect=Dialect.DUCKDB)[source]

Bases: object

Owns a scenario_id; orchestrates a multi-generator plant with compose-time collision detection + per-row metadata tagging on both base tables for surgical cleanup.

AV.5 — promoted from the spike. The spike used a side-table (<prefix>_scenario_claims) because daily_balances had no metadata column; AV.1 fixed that, so the production version tags PER ROW on both tables and drops the sidecar entirely.

prefix is the deployment’s table prefix (matches cfg.db_table_prefix). dialect selects the SQL JSON path helper. Defaults match the in-process SQLite test harness — every AS/AT/AU unit test that builds a generator + composes via ScenarioContext can pass nothing.

Parameters:

scenario_id (str)
prefix (str)
dialect (Dialect)

cleanup(conn)[source]

Delete every row on either base table whose metadata.scenario_id matches this scenario. Returns the total rowcount across both tables.

No sidecar table to consult — the tag IS the bookkeeping (AV.1 unlocked this). Other scenarios on overlapping accounts survive because their rows carry a different scenario_id.

Return type:: int
Parameters:: conn (SyncConnection)

compose(conn, *generators, dry_run=False)[source]

Pre-compose checks + emit all generators with the scenario_id threaded through + commit.

Two checks run before any emit fires:

Pairwise disjoint claims across the composed generators (catches “two DriftGenerators on the same role” at the wiring site, not at the DB-level PK violation).
Cross-scenario non-overlap against the data tables’ metadata.scenario_id (catches “scenario B overwrites scenario A’s rows” before any INSERT runs).

On success: each generator’s emit(conn, scenario_id=self.scenario_id) runs in order; the generator is responsible for tagging every row it writes.

AY.4.a — dry_run=True mode: conn must be a dry_run_capture(dialect) instance. The cross-scenario check is SKIPPED in dry-run (no real DB state exists to check against; the pairwise-disjoint check still fires since it’s a pure-data check on the generators themselves). Returns the captured [(sql, params), ...] list — pair with the AY.4.b renderer to produce static SQL text for the build_full_seed_sql path.

Live mode (default): returns None, commits the conn.

Return type:

list[tuple[str, tuple[object, ...]]] | None

Parameters:

conn (SyncConnection)
generators (ClaimedAccountsGenerator)
dry_run (bool)

dialect: Dialect = 'duckdb'

prefix: str = 'spec_example'

scenario_id: str

class recon_gen.common.spine.StuckPendingGenerator(transaction_id, transfer_id, rail_name, account_id, account_role, account_parent_role, max_pending_age_seconds, overshoot_seconds, as_of, prefix='spec_example')[source]

Bases: object

Emit a single Pending transaction whose posting is in the past of as_of by max_pending_age_seconds + overshoot_seconds. NO balance row, NO related Posted transactions — the stuck_pending matview reads only the transactions table, so the plant is single-row.

Post-AW.5: as_of is the owned temporal frame (matches what the matview reads from <prefix>_config.as_of). NO datetime.now() — plant + matview both read from one source; tests are deterministic; no TZ skew to absorb.

Parameters:

transaction_id (str)
transfer_id (str)
rail_name (str)
account_id (str)
account_role (str)
account_parent_role (str | None)
max_pending_age_seconds (int)
overshoot_seconds (int)
as_of (datetime)
prefix (str)

account_id: str

account_parent_role: str | None

account_role: str

as_of: datetime

property claimed_accounts: frozenset[str]: The single account_id this plant stucks a Pending tx on. AV.5.

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation

max_pending_age_seconds: int

overshoot_seconds: int

prefix: str = 'spec_example'

rail_name: str

transaction_id: str

transfer_id: str

class recon_gen.common.spine.StuckPendingInvariant(prefix='spec_example')[source]

Bases: object

Stuck-Pending detector. The matview gates on status = 'Pending' AND age_seconds > max_pending_age_seconds (per-rail cap from L2). Identity is (transaction_id, rail_name) — transaction_id alone is PK-unique, but rail_name rounds out the analyst-facing identity for diff readability.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'stuck_pending'

prefix: str = 'spec_example'

scenario_for(rail_name, *, as_of, overshoot_seconds=60, account_role='CustomerSubledger', instance=None, account_id=None)[source]

Resolve rail_name against the shape; return a generator that plants a Pending transaction on a account_role account with posting = as_of − (rail.max_pending_age + overshoot).

as_of is the owned temporal frame the matview reads from <prefix>_config.as_of (per AW.2). Caller is responsible for ensuring config.as_of matches — see common.l2.config_table.set_as_of. Tests pass a pinned datetime; production passes datetime.now() at scenario-creation time (same value the refresh helper UPDATEs into config).

overshoot_seconds=0 ⇒ age == threshold ⇒ matview’s > filter excludes ⇒ no fire (AP.2 non-violating convention adapted). Positive overshoot fires loud; negative also non-violating (age < threshold). Post-AW.5 the overshoots can be small natural values — no wall-clock skew to absorb.

Raises ValueError if the L2 has no rail with rail_name, OR if that rail has no max_pending_age set (stuck_pending’s matview filter excludes rails without one — manufacturing a scenario against an uncovered rail would silently emit an inert row, which we refuse).

AY.4.c — account_id overrides the default synthetic ID. The plant adapter (AY.4.c.3) threads OLD StuckPendingPlant.account_id through this kwarg so N plants on the same rail produce N distinct generators (the default f”acct-stuck-pending-{rail_name}” derivation would collide). Existing test callers can pass nothing → preserves the synthetic default byte-stable.

Return type:

StuckPendingGenerator

Parameters:

rail_name (str)
as_of (datetime)
overshoot_seconds (int)
account_role (str)
instance (L2Instance | None)
account_id (str | None)

class recon_gen.common.spine.StuckUnbundledGenerator(transaction_id, transfer_id, rail_name, account_id, account_role, account_parent_role, max_unbundled_age_seconds, overshoot_seconds, as_of, prefix='spec_example')[source]

Bases: object

Emit a single Posted transaction with bundle_id IS NULL whose posting is in the past of as_of by max_unbundled_age_seconds + overshoot_seconds. NO balance row, NO related rows.

Post-AW.5: as_of is the owned temporal frame; matview reads the same value from <prefix>_config.as_of → tests deterministic, no TZ skew.

Parameters:

transaction_id (str)
transfer_id (str)
rail_name (str)
account_id (str)
account_role (str)
account_parent_role (str | None)
max_unbundled_age_seconds (int)
overshoot_seconds (int)
as_of (datetime)
prefix (str)

account_id: str

account_parent_role: str | None

account_role: str

as_of: datetime

property claimed_accounts: frozenset[str]: The single account_id this plant strands as Posted-unbundled. AV.5.

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation

max_unbundled_age_seconds: int

overshoot_seconds: int

prefix: str = 'spec_example'

rail_name: str

transaction_id: str

transfer_id: str

class recon_gen.common.spine.StuckUnbundledInvariant(prefix='spec_example')[source]

Bases: object

Stuck-Unbundled detector. The matview gates on status = 'Posted' AND bundle_id IS NULL AND age_seconds > max_unbundled_age_seconds (per-rail cap from L2). Identity is (transaction_id, rail_name).

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'stuck_unbundled'

prefix: str = 'spec_example'

scenario_for(rail_name, *, as_of, overshoot_seconds=60, account_role='CustomerSubledger', instance=None, account_id=None)[source]

Resolve rail_name against the shape; plant a Posted-but- unbundled transaction with posting = as_of − (rail. max_unbundled_age + overshoot).

as_of is the owned temporal frame the matview reads from <prefix>_config.as_of (per AW.2). See StuckPendingInvariant .scenario_for for the full contract — same shape applies here.

Raises ValueError if rail doesn’t exist OR doesn’t have a max_unbundled_age (matview excludes those — uncovered scenario would silently inert).

AY.4.c — account_id overrides the default synthetic ID. The plant adapter (AY.4.c.3) threads OLD StuckUnbundledPlant.account_id through this kwarg so N plants on the same rail produce N distinct generators (the default f”acct-stuck-unbundled-{rail_name}” derivation would collide). Existing test callers can pass nothing → preserves the synthetic default byte-stable.

Return type:

StuckUnbundledGenerator

Parameters:

rail_name (str)
as_of (datetime)
overshoot_seconds (int)
account_role (str)
instance (L2Instance | None)
account_id (str | None)

class recon_gen.common.spine.SupersessionGenerator(account_id, account_role, account_scope, account_parent_role, rail_name, original_amount, corrected_amount, anchor_day, prefix='spec_example')[source]

Bases: object

Plant TWO transactions sharing one logical id — the original posting + a TechnicalCorrection rewrite.

Both rows land on the same account / rail / transfer_id; only the amount_money differs (the correction “fixes” the original amount). The dialect auto-increments entry so the correction sorts after the original; the audit PDF’s CASE on entry = MAX(entry) PARTITION BY id picks the correction as the “current” row + the original as the “superseded” trail.

Account context (role / scope / parent_role) arrives as construction args; the AY.4 adapter resolves them from the OLD plant’s referenced template instance.

Parameters:

account_id (str)
account_role (str)
account_scope (str)
account_parent_role (str | None)
rail_name (str)
original_amount (float)
corrected_amount (float)
anchor_day (date)
prefix (str)

account_id: str

account_parent_role: str | None

account_role: str

account_scope: str

anchor_day: date

property claimed_accounts: frozenset[str]

corrected_amount: float

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: AuditFixture: The audit-PDF supersession entry’s identity — keyed on the logical transaction id + the corrected amount (what the audit shows as the “current” value).

original_amount: float

prefix: str = 'spec_example'

rail_name: str

property transaction_id: str: The shared logical id for both rows — the supersession anchor. Deterministic on account_id.

property transfer_id: str

class recon_gen.common.spine.TrainingScenario(name, description, emitters, invariants, intended=<factory>, prefix='spec_example', instance_path=None)[source]

Bases: object

A docs-renderable, self-validating scenario.

Authors construct one of these per documented case; docs read name + description (free prose); the test suite calls self_validate(conn) to make sure the data the scenario emits actually produces the violations the prose claims.

intended is the load-bearing field — it’s the claim. A Violation in this set says: “the analyst will see this row in the matview after the scenario applies.” Implementation churn (different leg shapes, account IDs) is fine as long as the intended Violations still fire; that’s the same flexibility AS.5’s semantic_lock gives.

Parameters:

name (str)
description (str)
emitters (tuple[_Emitter, ...])
invariants (tuple[Invariant, ...])
intended (frozenset[Violation])
prefix (str)
instance_path (Path | None)

description: str

emitters: tuple[_Emitter, ...]

instance_path: Path | None = None

intended: frozenset[Violation]

invariants: tuple[Invariant, ...]

name: str

prefix: str = 'spec_example'

self_validate(conn)[source]

Apply the scenario; assert every intended Violation fires.

Raises AssertionError with the missing-Violation diff if the docs claim violations the data doesn’t produce. intended ⊆ detected is the contract — extra detected Violations (e.g., the secondary ledger_drift edge from a drift plant) are fine; missing claimed Violations are the failure mode.

Return type:: None
Parameters:: conn (SyncConnection)

class recon_gen.common.spine.Transfer(day, transfer_id, rail_name, legs, status='Posted', parent_transfer_id=None, origin='etl', hour=12)[source]

Bases: object

A multi-leg money-movement event sharing one transfer_id.

The double-entry invariant: sum(leg.amount for leg in legs) == 0 for a fully-balanced Posted transfer. NOT enforced at construction — Pending transfers commonly carry only one leg until they post; the matview SQL filters on Posted + matched legs, so an unbalanced transfer is harmless (it just doesn’t surface). The is_balanced helper exposes the check for callers that want it.

parent_transfer_id chains transfers into trails — money_trail’s matview walks this recursively (each Posted multi-leg transfer becomes one edge per chain hop).

Parameters:

day (date)
transfer_id (str)
rail_name (str)
legs (tuple[TransferLeg, ...])
status (Literal['Posted', 'Pending'])
parent_transfer_id (str | None)
origin (str)
hour (int)

day: date

hour: int = 12: Hour of day used for the posting timestamp on each leg. Defaults to noon so each leg lands inside the day’s balance window (00:00 ≤ posting < 24:00). Mirrors ts()’s convention.

is_balanced()[source]

True iff sum(leg.amount) == 0 — the double-entry conservation law. Used by callers (incl. tests) that want to verify a transfer is fully-resolved; the constructor doesn’t enforce it because Pending single-leg transfers are valid intermediate state.

Return type:: bool

legs: tuple[TransferLeg, ...]

origin: str = 'etl'

parent_transfer_id: str | None = None

rail_name: str

status: Literal['Posted', 'Pending'] = 'Posted'

transfer_id: str

class recon_gen.common.spine.TransferLeg(account_id, amount, account_name, account_role, account_scope, account_parent_role=None)[source]

Bases: object

One leg of a Transfer — money in or out of one account.

amount > 0 = money IN (Credit), < 0 = money OUT (Debit) — the project sign convention. The denormalized account-side fields (name, role, scope, parent_role) MUST be on each leg because _transactions is the source-of-truth table for the matviews; the matview SQL doesn’t JOIN to a separate account dimension.

Parameters:

account_id (str)
amount (float)
account_name (str)
account_role (str)
account_scope (Literal['internal', 'external'])
account_parent_role (str | None)

account_id: str

account_name: str

account_parent_role: str | None = None

account_role: str

account_scope: Literal['internal', 'external']

amount: float

class recon_gen.common.spine.TransferTemplateFactory(prefix='spec_example')[source]

Bases: object

Smart constructor namespace for TransferTemplateGenerator.

Resolves a TransferTemplate by name + its first leg_rail’s kind from the L2 instance, then builds a generator that plants one firing in the kind-appropriate shape.

Parameters:: prefix (str)

name: ClassVar[str] = 'transfer_template_firing'

prefix: str = 'spec_example'

scenario_for_template(template_name, *, source_account_id=None, destination_account_id=None, amount=100.0, firing_seq=1, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Resolve template_name to a TransferTemplate + the first leg_rail’s kind; build a generator that plants one firing.

Synthetic account defaults match RailFiringFactory’s shape: deterministic per-template strings when not supplied. Picker layer threads real materialized-account identifiers post-AY.4.

Raises ValueError when the template / leg_rail isn’t declared on the L2.

Return type:

TransferTemplateGenerator

Parameters:

template_name (str)
source_account_id (str | None)
destination_account_id (str | None)
amount (float)
firing_seq (int)
anchor_day (date)
instance (L2Instance | None)

class recon_gen.common.spine.TransferTemplateGenerator(template_name, rail_name, is_two_leg, source_account_id, destination_account_id, amount, single_leg_direction, firing_seq, anchor_day, prefix='spec_example')[source]

Bases: object

Emit one Posted firing of an L2-declared TransferTemplate.

Rail kind branches on is_two_leg (set by the factory from the template’s first leg_rail kind):

Two-leg: debit on source + credit on destination, both carrying template_name=template_name. Net = 0 (matches the template’s expected_net).

Single-leg: one leg on source in the resolved direction, carrying template_name=template_name. Net = ±amount (the L1 SQL surfaces as ‘Imbalanced’ against expected_net = 0 — accurate for a bare single-leg cycle).

intended returns a CoverageObservation keyed on (template_name, transfer_id, firing_seq). No matching Invariant.

Parameters:

template_name (str)
rail_name (str)
is_two_leg (bool)
source_account_id (str)
destination_account_id (str | None)
amount (float)
single_leg_direction (Literal['Debit', 'Credit'])
firing_seq (int)
anchor_day (date)
prefix (str)

amount: float

anchor_day: date

property claimed_accounts: frozenset[str]

destination_account_id: str | None

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

firing_seq: int

property intended: CoverageObservation

is_two_leg: bool

prefix: str = 'spec_example'

rail_name: str

single_leg_direction: Literal['Debit', 'Credit']

source_account_id: str

template_name: str

property transfer_id: str

class recon_gen.common.spine.TwoTemplateChainFactory(prefix='spec_example')[source]

Bases: object

Smart constructor namespace for TwoTemplateChainGenerator.

AY.2.b deliberately omits an Invariant here — there’s no matview to detect this plant’s evidence (the plant is non-violating). The factory mirrors the Invariant.scenario_for(…) pattern from the rest of the spine for surface parity, so seed-color generators feel like the violation-detecting ones at the call site.

Parameters:: prefix (str)

name: ClassVar[str] = 'two_template_chain_healthy'

prefix: str = 'spec_example'

scenario_for_healthy(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Pick any chain whose singleton child is a TransferTemplate (the AB.2.6 picker) and build a generator that plants a healthy 2-template chain firing.

Parent can resolve to either a Rail (rail-parent) or a TransferTemplate (template-parent, AG.3 Gap A); the generator carries enough fields to emit the parent row correctly in both cases.

Raises ValueError when the L2 has no eligible chain (singleton-child + parent ∈ rails ∪ templates + child ∈ templates with ≥1 leg_rail).

Return type:

TwoTemplateChainGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

class recon_gen.common.spine.TwoTemplateChainGenerator(chain_parent_name, parent_rail_name, parent_template_name, child_template_name, child_leg_rails, anchor_day, prefix='spec_example', account_id_override=None)[source]

Bases: object

Emit one parent leg + N child template legs (one per child_leg_rails entry) all sharing one child transfer_id and all carrying the same parent_transfer_id — the healthy 2-template chain shape.

Account fields are synthetic + deterministic (matches the ChainParentDisagreementGenerator pattern; the matview ignores account columns when grouping by transfer_id + template_name).

intended returns a CoverageObservation keyed on (transfer_id, chain_parent_name, child_template_name, child_leg_count) — the presence-claim tuple a coverage detector could round-trip against.

Parameters:

chain_parent_name (str)
parent_rail_name (str)
parent_template_name (str | None)
child_template_name (str)
child_leg_rails (tuple[str, ...])
anchor_day (date)
prefix (str)
account_id_override (str | None)

property account_id: str

account_id_override: str | None = None

anchor_day: date

chain_parent_name: str

child_leg_rails: tuple[str, ...]

child_template_name: str

property child_transfer_id: str

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: CoverageObservation

a healthy 2-template chain firing landed with N child legs all agreeing on parent_transfer_id. Identity carries the natural-key tuple a coverage detector would round-trip against if/when one lands.

Type:: Presence evidence

parent_rail_name: str

parent_template_name: str | None

property parent_transfer_id: str

prefix: str = 'spec_example'

class recon_gen.common.spine.Violation(invariant, identity)[source]

Bases: object

Abstract base for typed spine evidence.

Two fields:

invariant: the matview / coverage / fixture name (e.g. "drift", "rail_firing", "supersession"). The string discriminator the lock dict keys on; pairs with the runtime subtype to identify the specific shape.
identity: a frozenset of (column, value) pairs naming the row.

NEVER constructed directly in new code — use one of the three concrete subtypes (RuleViolation / CoverageObservation / AuditFixture). The base exists for typing + the legacy Violation.of(…) alias only.

Parameters:

invariant (str)
identity (frozenset[tuple[str, object]])

identity: frozenset[tuple[str, object]]

invariant: str

classmethod of(invariant, **identity)[source]

Backward-compat alias — returns a RuleViolation (a Violation subtype).

Pre-AY.2.a, every spine generator + detector built Violations via this constructor; AY.2.a layered in subtypes but kept the alias so the migration is incremental. New code should prefer the subtype-explicit constructors (RuleViolation.of(...) etc.) for type clarity.

Return type is the abstract Violation for liskov / covariant override compatibility — subclasses’ .of() can return their own concrete subtype without tripping pyright. At runtime this method returns a RuleViolation (the AS-era default).

Return type:

Violation

Parameters:

invariant (str)
identity (object)

class recon_gen.common.spine.ViolationGenerator(*args, **kwargs)[source]

Bases: Protocol

A producer of base-table rows intended to manifest intended.

intended is a @property (not a bare attribute) so concrete generators can derive it from their construction params — e.g., a DriftGenerator whose intended Violation includes its anchor day and resolved account_id, computed at access time.

AY.2.a — intended is typed Violation | None. The None case is the AP.2 non-violating shape: FanInChainGenerator(expected_ kind=’healthy’) plants parent_count = expected, so the fan_in_disagreement matview’s CASE produces no row, so the generator claims no Violation. AY.2.b’s CoverageObservation layering may flip the healthy variant to return a non-None CoverageObservation (signalling “I planted a chain firing”), but the Protocol stays widened to accommodate either shape.

emit(conn) writes the rows. Generators MAY also commit or refresh matviews internally, but most leave that to the caller so a single scenario can compose multiple generators against ONE connection and refresh once at the end (the AP.3 _assert_self_validates pattern).

emit(conn)[source]

Return type:: None
Parameters:: conn (SyncConnection)

property intended: Violation | None

class recon_gen.common.spine.XorGroupMissedFiringGenerator(template_name, xor_group_index, witness_rail_name, anchor_day, prefix='spec_example', account_id_override=None)[source]

Bases: object

Plant a Transfer tagged with template_name whose target XOR group fires NO members.

Emits ONE leg row carrying the template_name + a witness rail_name that is a leg_rail of the template but NOT in the target XOR group. The matview’s template_transfers CTE picks up the Transfer (template_name matches an XOR-grouped template); the LEFT JOIN against (transfer_id, template, member_rail) for the target group finds zero rows; firing_count = 0; the HAVING <> 1 gate surfaces the row with fired_rails = ‘’.

AY.4.c.2 — account_id_override allows the plant adapter (AY.4.c.3) to thread OLD plant account_ids through, preventing PK collisions when N plants of the same shape compose.

Parameters:

template_name (str)
xor_group_index (int)
witness_rail_name (str)
anchor_day (date)
prefix (str)
account_id_override (str | None)

property account_id: str: account_id_override wins when set (AY.4.c.2).

account_id_override: str | None = None

anchor_day: date

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation

prefix: str = 'spec_example'

template_name: str

property transfer_id: str

witness_rail_name: str

xor_group_index: int

class recon_gen.common.spine.XorGroupOverlapGenerator(template_name, xor_group_index, variant_a_rail_name, variant_b_rail_name, anchor_day, prefix='spec_example', account_id_override=None)[source]

Bases: object

Plant a Transfer tagged with template_name whose target XOR group fires TWO distinct members.

Emits TWO leg rows sharing one transfer_id + template_name, each carrying a different member rail_name from the target XOR group. The matview’s LEFT JOIN finds two member-rail firings for (transfer_id, template, target_group) → COUNT = 2 → HAVING <> 1 → row surfaces with fired_rails = ‘<a>,<b>’.

AY.4.c.2 — account_id_override allows the plant adapter (AY.4.c.3) to thread OLD plant account_ids through, preventing PK collisions when N plants of the same shape compose.

Parameters:

template_name (str)
xor_group_index (int)
variant_a_rail_name (str)
variant_b_rail_name (str)
anchor_day (date)
prefix (str)
account_id_override (str | None)

property account_id: str: account_id_override wins when set (AY.4.c.2).

account_id_override: str | None = None

anchor_day: date

property claimed_accounts: frozenset[str]

emit(conn, *, scenario_id=None)[source]

Return type:

None

Parameters:

conn (SyncConnection)
scenario_id (str | None)

property intended: RuleViolation

prefix: str = 'spec_example'

template_name: str

property transfer_id: str

variant_a_rail_name: str

variant_b_rail_name: str

xor_group_index: int

class recon_gen.common.spine.XorGroupViolationInvariant(prefix='spec_example')[source]

Bases: object

Detector for the AB.3.3 matview.

Identity tuple: (transfer_id, template_name, xor_group_index). The matview’s other columns (firing_count, fired_rails, business_day) are diagnostic — they distinguish missed (0) from overlap (≥2) but the identity-level “did this XOR group misfire” is what the spine carries.

Parameters:: prefix (str)

detect(conn)[source]

Return type:: set[Violation]
Parameters:: conn (SyncConnection)

name: ClassVar[str] = 'xor_group_violation'

prefix: str = 'spec_example'

scenario_for_missed(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Pick an XOR-grouped template with ≥1 leg_rail outside the target group (the witness rail). Returns a generator that plants ONE Transfer firing the witness — firing_count=0 for the target group, matview surfaces a row.

Raises ValueError if no template has both an XOR group AND a witness leg_rail outside it (the AB.3.5 picker’s input requirement).

Return type:

XorGroupMissedFiringGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

scenario_for_overlap(*, anchor_day=datetime.date(2030, 1, 1), instance=None)[source]

Pick an XOR group with ≥2 members (validator C1d enforces the ≥2 condition at load time). Returns a generator that plants ONE Transfer firing TWO members of the same XOR group — firing_count=2, matview surfaces a row.

Raises ValueError if no template declares any leg_rail_xor_groups.

Return type:

XorGroupOverlapGenerator

Parameters:

anchor_day (date)
instance (L2Instance | None)

recon_gen.common.spine.apply_scenario(conn, *emitters, prefix='spec_example', instance_path=None, dialect=Dialect.DUCKDB)[source]

Emit every passed object’s rows into conn, commit, then refresh the matviews for the prefix’s L2 instance.

instance_path=None uses the bundled tests/l2/spec_example.yaml — same default the existing in-process spike harness pattern uses. Pass an explicit path for AT’s Investigation-surface scenarios.

dialect defaults to Dialect.DUCKDB for the in-process test harness shape (the dominant call site). AY.3 lifted the hardcode so production callers + per-dialect semantic_lock generation (AZ.1 will produce _semantic_locks/<instance>.<dialect>.json snapshots that need PG + Oracle paths through this function) can thread the deployed dialect through. The spine emitters already detect the dbapi placeholder style per-connection (AT.5.b’s _emit_helpers._placeholder_style); this kwarg covers the matview refresh + script execution leg of the same path.

Return type:

None

Parameters:

conn (SyncConnection)
emitters (_Emitter)
prefix (str)
instance_path (Path | None)
dialect (Dialect)

recon_gen.common.spine.dry_run_capture(dialect=Dialect.DUCKDB)[source]

Return a fresh dry-run capture conn for the given dialect.

The returned object satisfies the dbapi 2.0 shape insert_tx / insert_balance rely on: cursor() → cursor with execute(sql, params), commit() / close() no-op. captured accumulates every emitted (sql, params) pair in arrival order.

Pair with ScenarioContext.compose(conn=dry_run_capture(d), …, dry_run=True) (AY.4.a) + the AY.4.b renderer.

Return type:: _DryRunBase
Parameters:: dialect (Dialect)

recon_gen.common.spine.generators_for(invariant_class)[source]

The generator classes whose emission trips this invariant. Reverse-lookup over the edge table.

Return type:: set[type[ViolationGenerator]]
Parameters:: invariant_class (type[Invariant])

recon_gen.common.spine.invariants_for(generator_class)[source]

The invariants a given generator’s emission trips. Empty tuple when the generator class isn’t registered (yet).

Return type:: tuple[type[Invariant], ...]
Parameters:: generator_class (type[ViolationGenerator])

recon_gen.common.spine.iter_edges()[source]

Every (generator_class, invariant_class) edge as a flat sequence. Convenient for parametrized property tests.

Return type:: Iterator[tuple[type[ViolationGenerator], type[Invariant]]]

recon_gen.common.spine.lock_to_json(lock, *, instance, dialect, canonical_anchor)[source]

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.

Parameters:

lock (dict[str, frozenset[Violation]]) – dict[invariant_name, frozenset[Violation]] — exactly what semantic_lock(conn, ALL_INVARIANTS) returns.
instance (str) – the L2 instance name (e.g., “spec_example”) — lands in scenario_fingerprint.instance for mismatch detection.
dialect (Dialect) – the SQL dialect — lands in scenario_fingerprint.dialect.
canonical_anchor (date) – the canonical lock date — lands in scenario_fingerprint.canonical_anchor.

Return type:

str

Returns:

A JSON string with indent=2, terminating newline. Suitable for direct write-to-disk + byte-equality test.

recon_gen.common.spine.render_captured_sql(captured, *, dialect, statement_separator=';\\n')[source]

Walk captured (sql, params) pairs + render each as a static SQL statement. Returns the concatenated script with each statement terminated by statement_separator (default ;n).

Use after ScenarioContext.compose(dry_run=True) to feed the output to emit_to_target / write-to-disk / pipe-to-psql.

Return type:

str

Parameters:

captured (Iterable[tuple[str, tuple[object, ...]]])
dialect (Dialect)
statement_separator (str)

recon_gen.common.spine.scenario_metadata(scenario_id, **extra)[source]

Build the JSON metadata blob a tagged plant row carries.

Centralized so the cleanup query can extract via a portable JSON_VALUE(metadata, '$.scenario_id') and every tagging caller writes the same shape. Extra kwargs round-trip — useful for per-generator attribution (e.g. generator='DriftGenerator').

Return type:

str

Parameters:

scenario_id (str)
extra (object)

recon_gen.common.spine.scenario_rng(seed=None)[source]

Build a seeded random.Random for a ViolationGenerator.

seed=None uses SCENARIO_BASE_SEED — every scenario_for call site that doesn’t pin its own seed lands at the same starting point, so the locked-seed agreement (tests/data/test_locked_seeds.py shape) keeps holding.

Return type:: Random
Parameters:: seed (int | None)

recon_gen.common.spine.scenario_to_generators(scenarios, instance, *, frame=None, prefix='spec_example')[source]

Walk every plant collection on scenarios + return the matching spine generator per plant. Order matches the OLD emit_seed’s per-kind dispatch (drift → overdraft → limit → inbound cap → two-template → … → inv-fanout) so debug output stays diff-friendly.

BD.3 (2026-05-25): the only temporal input is frame: AsOfFrame. The legacy anchor / plant_window / as_of kwargs were dropped (no-compat-shim — pre-stable framing). The four production callers — cli/data.py::data apply, common/l2/seed.py × 2, tests/e2e/_seed_helpers.py — each construct a frame at the seam.

Frame resolution:

frame is supplied → anchor_day = frame.as_of, plant_window = frame.window.
Not supplied (positional-only test callers in tests/unit/test_spine_ay4c3_plant_adapter.py and other bare emit_seed(inst, scenario) test seams) → degenerate fallback frame: as_of = scenarios.today, plant_window = [scenarios.today - 90, scenarios.today]. The 90-day backward window matches emit_baseline_seed’s default window_days=90 so plants with days_ago up to 90 fit the at_offset_from_end validation. Single-day windows break L1 plants with days_ago > 0 (drift/overdraft default days_ago=4).

wall_clock (for StuckPending / StuckUnbundled generators) derives from anchor_day at noon.

prefix defaults to “spec_example” (the in-process test harness shape). Production callers (AY.4.d build_full_seed_sql) pass cfg.db_table_prefix; every constructed generator inherits self.prefix = prefix so insert_tx / insert_balance writes to the correctly-prefixed tables.

Return type:

tuple[ViolationGenerator, ...]

Parameters:

scenarios (ScenarioPlant)
instance (L2Instance)
frame (AsOfFrame | None)
prefix (str)

recon_gen.common.spine.semantic_lock(conn, invariants)[source]

Run detect(conn) for every invariant; return the lock dict.

Returned dict is {invariant.name: frozenset(detected_violations)}. frozenset makes the value byte-stable: two runs that produce the same set in different insertion orders compare equal.

The DICT itself is not frozen, but its keys + values are immutable; equality (lock_a == lock_b) is the gate. The caller decides whether to lock against a literal Python value, a JSON-serialized snapshot, or a tests/data/_semantic_locks/<scenario>.json file (the eventual successor to _locked_seeds).

Return type:

dict[str, frozenset[Violation]]

Parameters:

conn (SyncConnection)
invariants (Iterable[Invariant])

recon_gen.common.spine.validate_all(scenarios, conn_factory)[source]

Validate a batch of scenarios, each against its OWN fresh DB.

conn_factory is a no-arg callable returning a fresh sqlite3. Connection with the schema already applied (the in-process harness pattern). Each scenario gets its own connection so prior emissions don’t bleed into the next test’s detect set.

Useful in a docs-build hook: collect every registered TrainingScenario and validate them all in one shot before rendering the prose. A failure halts the build with the missing- violation diff.

Return type:

None

Parameters:

scenarios (Iterable[TrainingScenario])
conn_factory (Callable[[], SyncConnection])

Modules

`account_simulation`	AccountSimulation — the AP.2 stateful-fold pattern, productionized.
`anomaly`	Anomaly family — windowed-statistical L2 invariant + generator.
`anomaly_view`	Anomaly View — owns the σ-threshold (AP.3 finding #3 lock).
`chain_completion`	AY.4.g — ChainCompletionGenerator (chain-completion shim).
`chain_parent_disagreement`	Chain-parent-disagreement family — Invariant + ViolationGenerator.
`drift`	Drift family — concrete Invariant + ViolationGenerator impls.
`dry_run_renderer`	AY.4.b — render captured (sql, params) pairs as static SQL text.
`expected_eod`	Expected-EOD-balance family — Invariant + ViolationGenerator.
`failed_transaction`	FailedTransaction family — spine generator only (no Invariant).
`fan_in_disagreement`	Fan-in-disagreement family — Invariant + ViolationGenerator.
`generator`	ViolationGenerator Protocol — the producer.
`inv_fanout`	Investigation fanout (seed-color) family — ViolationGenerator only.
`invariant`	Invariant Protocol — the rule + detector.
`ledger_simulation`	LedgerSimulation — vector-state composition of AccountSimulation`s plus the AT.3 `Transfer primitive for cross-account flow.
`limit_breach`	Limit-Breach family — deepest L2 coupling of the L1 spine.
`money_trail`	Money-trail family — recursive-graph L2 invariant + chain generator.
`multi_xor_violation`	Multi-XOR-violation family — Invariant + 2 `ViolationGenerator`s.
`overdraft`	Overdraft family — concrete Invariant + ViolationGenerator impls.
`plant_adapter`	AY.4.c.3 — scenario_to_generators(plants, instance, anchor).
`rail_firing`	Rail firing (broad-mode) family — ViolationGenerator only.
`registry`	Invariant ⋈ generator registry — many-to-many edge bookkeeping.
`rng`	Deterministic RNG factory for ViolationGenerator impls.
`scenario_context`	ScenarioContext — composition safety + cleanup attribution (AV.5).
`semantic_lock`(conn, invariants)	Run detect(conn) for every invariant; return the lock dict.
`semantic_lock_json`	AZ.1 — JSON serialization for the semantic lock dict.
`stuck_pending`	Stuck-Pending family — first transaction-based + instance-coupled spine invariant.
`stuck_unbundled`	Stuck-Unbundled family — twin of stuck_pending with disjoint conditions.
`supersession`	Supersession family — spine generator only (no Invariant).
`training`	Training/docs scenarios that self-validate.
`transfer_template`	TransferTemplate firing (broad-mode) family — ViolationGenerator only.
`two_template_chain`	Two-template chain (healthy) family — ViolationGenerator only.
`violation`	Violation — the typed evidence currency of the spine.
`xor_group_violation`	XOR-group-violation family — Invariant + 2 `ViolationGenerator`s.