"""Stuck-Pending family — first transaction-based + instance-coupled
spine invariant.
`StuckPendingInvariant` fires when a Pending transaction's posting age
exceeds the rail's configured `max_pending_age`. This is the first
spine invariant that:
1. **Reads transactions, not balances** — every prior promotion
(DriftGenerator, OverdraftGenerator, ExpectedEodBalanceGenerator)
emits daily_balances rows. StuckPendingGenerator emits a Pending
transaction with no balance row.
2. **Couples to the L2 instance shape** — `scenario_for(rail_name)`
reads the rail's `max_pending_age` from L2 to plant a transaction
whose age overshoots it. The shape-resolution discipline AS.2's
`DriftInvariant.scenario_for(role)` started extends here: the
invariant owns L2 resolution, fails loud at the request site.
3. **Uses wall-clock time** — the matview computes `age_seconds =
CURRENT_TIMESTAMP - posting`. The plant computes posting via
`datetime.now() - (max_pending_age + overshoot)` so that at refresh
time, age_seconds > max_pending_age_seconds → fires. Identity is
`(transaction_id, rail_name)` — stable across refreshes; age value
itself is NOT in the identity (it drifts with wall-clock).
Per AU.0/AU.2 lessons: the empirical edge check is mandatory. Prediction
(written before the test): stuck_pending trips ONLY itself. Pending
transactions don't contribute to drift's computed_subledger_balance
(filters status='Posted'); no balance row emitted ⇒ no overdraft, no
expected_eod, no ledger_drift, no drift. Single-edge registry entry
expected. The test verifies empirically — if it surfaces an unexpected
edge, the AU.x cadence's stop-and-evaluate catches it.
`overshoot_seconds=0` is the non-violating shape (age == threshold;
matview's `>` filter excludes). Same AP.2 convention adapted to the
seconds-unit knob.
"""
from __future__ import annotations
from recon_gen.common.db import SyncConnection
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import ClassVar
from recon_gen.common.l2.primitives import L2Instance, SingleLegRail, TwoLegRail
from recon_gen.common.spine._db import fetch_all
from recon_gen.common.spine._emit_helpers import (
find_internal_with_role,
insert_tx,
load_spec_example,
)
from recon_gen.common.spine.violation import RuleViolation, Violation
# Either rail subtype is acceptable — both carry `max_pending_age`.
_RailWithPendingAge = TwoLegRail | SingleLegRail
[docs]
@dataclass(frozen=True)
class StuckPendingInvariant:
"""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."""
name: ClassVar[str] = "stuck_pending"
prefix: str = "spec_example"
[docs]
def detect(self, conn: SyncConnection) -> set[Violation]:
rows = fetch_all(
conn,
f"SELECT transaction_id, rail_name "
f"FROM {self.prefix}_stuck_pending",
)
return {
RuleViolation.of(
"stuck_pending",
transaction_id=str(tid),
rail_name=str(rn),
)
for tid, rn in rows
}
[docs]
def scenario_for(
self,
rail_name: str,
*,
as_of: datetime,
overshoot_seconds: int = 60,
account_role: str = "CustomerSubledger",
instance: L2Instance | None = None,
account_id: str | None = None,
) -> "StuckPendingGenerator":
"""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.
"""
inst = instance if instance is not None else load_spec_example()
rail = _find_rail_with_max_pending_age(inst, rail_name)
# `_find_rail_with_max_pending_age` raises if `max_pending_age`
# is None — pyright doesn't narrow the union member, so help it.
assert rail.max_pending_age is not None
acct = find_internal_with_role(
inst, account_role, error_kind="stuck_pending",
)
resolved_account_id = (
account_id or f"acct-stuck-pending-{rail_name}"
)
# AY.4.c.4 — fold account_id into transaction_id / transfer_id so
# N plants on the same rail (different accounts) don't PK-collide
# at INSERT time. Default-case account_id derivation IS rail-keyed,
# so single-test callers stay byte-stable.
return StuckPendingGenerator(
transaction_id=f"tx-stuck-pending-{rail_name}-{resolved_account_id}",
transfer_id=f"xfer-stuck-pending-{rail_name}-{resolved_account_id}",
rail_name=rail_name,
account_id=resolved_account_id,
account_role=account_role,
account_parent_role=acct.parent_role,
max_pending_age_seconds=int(rail.max_pending_age.total_seconds()),
overshoot_seconds=overshoot_seconds,
as_of=as_of,
)
[docs]
@dataclass
class StuckPendingGenerator:
"""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.
"""
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
# AY.4.d — production callers thread cfg.db_table_prefix here.
prefix: str = "spec_example"
@property
def intended(self) -> RuleViolation:
return RuleViolation.of(
"stuck_pending",
transaction_id=self.transaction_id,
rail_name=self.rail_name,
)
@property
def claimed_accounts(self) -> frozenset[str]:
"""The single account_id this plant stucks a Pending tx on. AV.5."""
return frozenset({self.account_id})
[docs]
def emit(
self,
conn: SyncConnection,
*,
scenario_id: str | None = None,
) -> None:
from recon_gen.common.spine.scenario_context import scenario_metadata
metadata = (
scenario_metadata(scenario_id, generator="StuckPendingGenerator")
if scenario_id is not None else None
)
# Plant `posting` far enough in the past of `as_of` that the
# matview's `age_seconds > max_pending_age_seconds` filter fires.
# Use a Credit posting (sign-direction CHECK constraint requires
# money>=0 for Credit; arbitrary positive value).
age_back = self.max_pending_age_seconds + self.overshoot_seconds
posting_dt = self.as_of - timedelta(seconds=age_back)
insert_tx(
conn,
prefix=self.prefix,
id=self.transaction_id,
account_id=self.account_id,
account_name=f"Stuck Pending ({self.rail_name})",
account_role=self.account_role,
account_scope="internal",
account_parent_role=self.account_parent_role,
amount_money=100.0,
amount_direction="Credit",
status="Pending",
posting=posting_dt.strftime("%Y-%m-%d %H:%M:%S"),
transfer_id=self.transfer_id,
rail_name=self.rail_name,
origin="etl",
metadata=metadata,
)
# ---------------------------------------------------------------------------
# Stuck-pending-specific rail finder — per-invariant-shape, no
# duplication burden. Shared helpers live in
# `common/spine/_emit_helpers.py` post-AU.3.d.
# ---------------------------------------------------------------------------
def _find_rail_with_max_pending_age(
instance: L2Instance, rail_name: str,
) -> _RailWithPendingAge:
"""Return the L2 rail with the given `name` AND a non-None
`max_pending_age`. Raises ValueError if either condition fails —
the matview excludes rails without `max_pending_age` (the JOIN to
`<prefix>_config.l2_yaml`'s rails iteration filters NULLs out via
the outer WHERE), so a scenario against an uncovered rail would
silently inert; we refuse instead.
Returns the concrete rail type (TwoLegRail | SingleLegRail) so the
caller's `.max_pending_age.total_seconds()` typechecks without
runtime introspection."""
for r in instance.rails:
if r.name == rail_name:
if r.max_pending_age is None:
raise ValueError(
f"rail {rail_name!r} has no max_pending_age set; "
f"stuck_pending's matview excludes it. Cannot "
f"manufacture a stuck_pending scenario against this rail."
)
return r
raise ValueError(
f"shape has no rail named {rail_name!r}; cannot manufacture "
f"a stuck_pending scenario"
)