"""QuickSight DataSet builders for the L2 Flow Tracing app.
The Chains / L2 Exceptions tabs join L2-declared values (static, from
the L2 instance) to runtime activity (from the prefixed
``<prefix>_current_transactions`` matview). The L2 declarations are
inlined into the SQL as a CTE of literal rows — no per-rail dataset
proliferation, no per-instance database table.
The Rails tab is a transactions explorer (M.3.10c rewrite — the
M.3.5 declared-rails table moves to a future Docs tab). It uses two
new datasets that participate in the metadata cascade:
- ``l2ft-postings-ds``: one row per leg, parameterized on ``pKey`` +
``pValues`` so the metadata cascade filters it via QS ``<<$param>>``
substitution into a JSONPath.
- ``l2ft-meta-values-ds``: distinct metadata values for the chosen
key, parameterized on ``pKey`` so the Value dropdown narrows when
the Key dropdown changes.
Substep landmarks:
- M.3.4 — skeleton (no datasets)
- M.3.5 — Rails dataset (later DROPPED in M.3.10c — moves to Docs tab)
- M.3.6 — Chains dataset
- M.3.7 — L2 Exceptions datasets (six small KPI-backers)
- M.3.8 — Auto metadata-driven filter dropdown sources
(later DROPPED in M.3.10c — replaced by the cascade)
- M.3.10c — Rails tab redesign on dataset parameters
"""
from __future__ import annotations
from datetime import timedelta
from recon_gen.common.config import Config
from recon_gen.common.dataset_contract import (
ColumnShape,
ColumnSpec,
DatasetContract,
build_dataset,
)
from recon_gen.common.l2 import (
L2Instance,
RoleExpression,
SingleLegRail,
TwoLegRail,
posted_requirements_for,
)
from recon_gen.common.models import (
DataSet,
DatasetParameter,
DateTimeDatasetParameter,
DateTimeDatasetParameterDefaultValues,
StringDatasetParameter,
StringDatasetParameterDefaultValues,
)
from recon_gen.common.sheets.app_info import (
build_liveness_dataset,
build_matview_status_dataset,
)
from recon_gen.common.sql import (
Dialect,
dual_from,
greatest,
json_value,
typed_null,
universal_date_range_clause,
)
from recon_gen.common.sql.money import cents_to_dollars_sql
[docs]
def l2ft_matview_specs(
cfg: Config,
) -> list[tuple[str, str | None]]:
"""Matviews + base tables the L2 Flow Tracing dashboard reads,
paired with the date column for App Info's ``latest_date`` KPI.
Includes the base tables (transactions / daily_balances) so the
operator can spot stale matviews against fresh ETL loads at a
glance. Mirrors ``l1_matview_specs`` / ``inv_matview_specs``.
"""
p = cfg.db_table_prefix
return [
(f"{p}_transactions", "posting"),
(f"{p}_daily_balances", "business_day_start"),
(f"{p}_current_transactions", "posting"),
(f"{p}_current_daily_balances", "business_day_start"),
]
# Phase BM — per-sheet date-range filter param names. Each L2FT
# sheet (Rails / Chains / Transfer Templates) carries its own pair so
# the analyst's window on one sheet doesn't perturb another. Pre-BM
# these were analysis-only (declared in ``app.py``) and narrowed via
# per-sheet ``TimeRangeFilter`` FilterGroups on the QS side; App2
# saw the picker widget but didn't actually narrow (no
# ``{date_filter}`` slot in the L2FT dataset SQL). BM unifies via
# dataset-SQL pushdown — both renderers narrow at the DB.
P_L2FT_RAILS_DATE_START = "pL2ftDateStart"
P_L2FT_RAILS_DATE_END = "pL2ftDateEnd"
P_L2FT_CHAINS_DATE_START = "pL2ftChainsDateStart"
P_L2FT_CHAINS_DATE_END = "pL2ftChainsDateEnd"
P_L2FT_TT_DATE_START = "pL2ftTtDateStart"
P_L2FT_TT_DATE_END = "pL2ftTtDateEnd"
# Match-all sentinel dates — the L2FT default state shows ALL data
# (operator narrows down from there). Differs from L1 / Exec which
# default to a 7-day / 30-day window centered on the as-of anchor —
# L2FT is a triage surface, not a daily-operations dashboard, so the
# full window is the right default. The ``T00:00:00`` form is the
# canonical ISO-8601 shape ``universal_date_range_clause`` expects on
# both renderers (QS substitutes the literal; App2 binds the string).
_L2FT_DATE_START_STATIC = "1900-01-01T00:00:00"
_L2FT_DATE_END_STATIC = "2099-12-31T00:00:00"
def _l2ft_match_all_range_params(
start_param: str, end_param: str,
) -> list[DatasetParameter]:
"""Phase BM — the two ``DateTimeDatasetParameter``s for an L2FT
date-range pickup pair. Defaults are the match-all sentinels.
``app.py``'s analysis-level ``DateTimeParam`` declarations bridge
into the same names via ``MappedDataSetParameters``.
"""
return [
DatasetParameter(DateTimeDatasetParameter=DateTimeDatasetParameter(
Name=start_param, ValueType="SINGLE_VALUED",
TimeGranularity="DAY",
DefaultValues=DateTimeDatasetParameterDefaultValues(
StaticValues=[_L2FT_DATE_START_STATIC],
),
)),
DatasetParameter(DateTimeDatasetParameter=DateTimeDatasetParameter(
Name=end_param, ValueType="SINGLE_VALUED",
TimeGranularity="DAY",
DefaultValues=DateTimeDatasetParameterDefaultValues(
StaticValues=[_L2FT_DATE_END_STATIC],
),
)),
]
def _l2ft_date_range_clause(
date_column: str, start_param: str, end_param: str, cfg: Config,
) -> str:
"""Phase BM — day-inclusive predicate fragment narrowing
``date_column`` by the supplied ``start_param`` / ``end_param``
dataset-param names. Thin wrapper over
:func:`universal_date_range_clause`."""
return universal_date_range_clause(
date_column,
start_param=start_param,
end_param=end_param,
dialect=cfg.dialect,
)
# Visual identifiers — keys for the Dataset registry on App.
DS_POSTINGS = "l2ft-postings-ds"
DS_META_VALUES = "l2ft-meta-values-ds"
DS_CHAINS = "l2ft-chains-ds"
DS_CHAIN_INSTANCES = "l2ft-chain-instances-ds"
DS_TT_INSTANCES = "l2ft-tt-instances-ds"
DS_TT_LEGS = "l2ft-tt-legs-ds"
DS_UNIFIED_L2_EXCEPTIONS = "l2ft-unified-exceptions-ds"
# M.3.7 — six L2 exception sections, each backed by its own narrow dataset.
DS_EXC_CHAIN_ORPHANS = "l2ft-exc-chain-orphans-ds"
DS_EXC_UNMATCHED_RAIL_NAME = "l2ft-exc-unmatched-rail-name-ds"
DS_EXC_DEAD_RAILS = "l2ft-exc-dead-rails-ds"
DS_EXC_DEAD_BUNDLES_ACTIVITY = "l2ft-exc-dead-bundles-activity-ds"
DS_EXC_DEAD_METADATA = "l2ft-exc-dead-metadata-ds"
DS_EXC_DEAD_LIMIT_SCHEDULES = "l2ft-exc-dead-limit-schedules-ds"
# Sentinel value for the metadata Key parameter's default. The
# transactions dataset's WHERE clause short-circuits to "no metadata
# filter" when the picked key equals this sentinel, so a freshly-
# loaded dashboard renders all rows even before the analyst engages
# the cascade. Placed at module scope so app.py + tests can reference
# it from a single source of truth.
META_KEY_ALL_SENTINEL = "__ALL__"
# Sentinel default for the multi-valued Value parameter. When the
# Key has been picked but no Value has yet been chosen, this default
# matches no real metadata value → the table goes empty as a hint
# the analyst still needs to pick a Value. Lives in CamelCase-safe
# form (no underscores other than at the boundaries — QS parameter
# *names* require alphanumerics, but parameter *values* can be
# anything the SQL accepts).
META_VALUE_PLACEHOLDER_SENTINEL = "__placeholder__"
# Y.2.c/d — "match no real value" sentinel — ``col IN ('__no_match__')``
# is valid SQL returning zero rows, the right outcome for a SINGLE_VALUED
# dropdown that should start empty (Account Network anchor) or a
# completion-status / bundle-status enum the L2 happens to have no rows
# for. (Pre-X.2.t.2 it was also the fallback for an empty declared-value
# list on a MULTI_VALUED enum default; those now default to
# ``[L2FT_ALL_SENTINEL]``, never an empty list.)
PUSHDOWN_NO_MATCH_SENTINEL = "__no_match__"
# X.2.t.2 + AA.A.3 — "match everything on load" sentinel — the static
# default for every SINGLE_VALUED pushdown dataset param post-AA.A.3.
# Pre-AA.A.3 the pushdown params were MULTI_VALUED with a sentinel
# 1-element-list default (the X.2.t.2 cap dodge for the rail / chain /
# template name dropdowns — an institution may declare >32 of any of
# them, and AWS caps ``StringDatasetParameter.DefaultValues.StaticValues``
# at 32). AA.A.3 flipped to SINGLE_VALUED per the drill-to-one default
# (operator workflow on these dashboards is pick-one-value 99% of the
# time; the MULTI shape had no one-click "pick this value and clear the
# rest" gesture — see ``docs/audits/aa_a_dropdown_audit.md``).
#
# The SQL guards ``('__l2ft_all__' = <<$pX>> OR col = <<$pX>>)`` (see
# ``_match_all_in_clause``): on load (and when the dropdown is cleared,
# reverting the dataset param to this default) the first disjunct is
# true so all rows pass; the bridge then maps the control's picked value
# in and the second disjunct narrows to that single value. The control's
# *options* still come from the full declared list — AWS doesn't cap that.
# Fixed-schema enums (``status`` / ``bundle_status`` /
# ``completion_status``, ≤ 3 elements by construction) also flipped to
# the same SINGLE shape per the AA.A.3 universal flip.
L2FT_ALL_SENTINEL = "__l2ft_all__"
_L2FT_ALL_SENTINEL_SQL = f"'{L2FT_ALL_SENTINEL}'"
def _match_all_in_clause(col: str, param_name: str) -> str:
"""WHERE-fragment for a SINGLE_VALUED pushdown dropdown whose dataset
param defaults to the bare ``L2FT_ALL_SENTINEL``: ``('__l2ft_all__' =
<<$p>> OR col = <<$p>>)``. Mirrors ``apps/l1_dashboard``'s
``_data_value_clause``.
AA.A.3 collapsed the prior multi-value form (``IN (...)``) into this
single-value scalar form when drill-to-one became the default
operator workflow (audit at ``docs/audits/aa_a_dropdown_audit.md``);
the function name carries over from the X.2.t.2 multi shape — at
call sites it still reads as "match-all guard for this column +
param", which is the load-time semantic regardless of the predicate
operator."""
return (
f"({_L2FT_ALL_SENTINEL_SQL} = <<${param_name}>>"
f" OR {col} = <<${param_name}>>)"
)
# AK.1 — dataset-parameter Ids are assigned in build_dataset via
# auto_id(f"{dataset_id}:dsparam:{Name}"); construction sites below leave
# Id unset. The old hand-picked GUID-shaped constants (11111111…) were
# IDENTICAL across datasets sharing a param name (pKey lives on postings +
# chain-instances + tt-legs), which collides when an analysis spans them —
# QS rejects the whole analysis. Deriving per (dataset, name) keeps emit
# byte-stable AND unique.
# Per-Chain-row-child edge — declared parent→child relationship +
# runtime parent firing counts + matched-child counts + orphan rate.
# A row IS one Sankey edge in the Chains visual; a multi-children
# Chain row contributes N rows.
#
# M.3.10d: no longer wired into the Chains sheet (the Sankey + edge
# details moved out in favor of a per-instance explorer); kept in the
# module for the M.7 Docs render of declared topology.
CHAINS_CONTRACT = DatasetContract(columns=[
ColumnSpec("parent_name", "STRING"),
ColumnSpec("child_name", "STRING"),
ColumnSpec("required", "STRING"), # 'Required' / 'Optional' for display
ColumnSpec("xor_group", "STRING"), # NULL when no XOR membership
ColumnSpec("source_node", "STRING"), # display label for the parent node
ColumnSpec("target_node", "STRING"), # display label for the child node
ColumnSpec("parent_firing_count", "INTEGER"),
ColumnSpec("child_firing_count", "INTEGER"),
ColumnSpec("orphan_count", "INTEGER"),
ColumnSpec("orphan_rate", "DECIMAL"),
])
# Per-parent-firing chain-instance row backing the Chains sheet's
# explorer (M.3.10d). One row per distinct parent transfer firing of
# any L2-declared chain-parent name; ``completion_status`` is computed
# inline from required-child firings against the parent's transfer_id.
# Parameterized on pKey + pValues for the metadata cascade.
CHAIN_INSTANCES_CONTRACT = DatasetContract(columns=[
# parent_chain_name is a drill destination for the L2 Exceptions
# table's "View in Chains" right-click — see
# UNIFIED_L2_EXCEPTIONS_CONTRACT for the full drill story. Holds
# either a rail OR a template name per SPEC.
ColumnSpec(
"parent_chain_name", "STRING",
shape=ColumnShape.L2_DECLARED_NAME,
),
ColumnSpec("parent_transfer_id", "STRING"),
ColumnSpec("parent_posting", "DATETIME"),
ColumnSpec("parent_status", "STRING"),
ColumnSpec("parent_amount_money", "DECIMAL"),
# BK.7 — display names disambiguate "Required" at the chain-grammar
# level (a child the L2 declares as MUST FOLLOW the parent) from the
# operator's natural read "legs of a multi-leg transfer". A
# singleton-children chain row is marked Required by the
# `_declared_chains_cte` (Z.A: "always fires when parent fires");
# multi-children chain rows are XOR siblings and ALL marked
# Optional, so required_total=0 there is the correct count.
# ``required_fired`` is the per-firing count of those declared
# Required children that actually fired against the parent's
# transfer_id. Pre-BK.7 the auto-titled headers ("Required Total" /
# "Required Fired") read as transfer-leg counts; the rename makes
# the chain-grammar scope explicit.
ColumnSpec(
"required_total", "INTEGER",
display_name="Required Children Declared",
),
ColumnSpec(
"required_fired", "INTEGER",
display_name="Required Children Fired",
),
ColumnSpec("completion_status", "STRING"),
])
# Per-shared-Transfer row backing the Transfer Templates sheet's
# Table (M.3.10f, completion_status reshaped in M.3.10j).
#
# ``completion_status`` (M.3.10j) combines the L1 conservation check
# (``actual_net`` ≈ ``expected_net``) with the L2 chain completeness
# check (every Required child fired AND every XOR group has exactly
# one fired). Three states cover the meaningful outcomes:
#
# - 'Complete' — balanced AND no chain orphans / XOR violations
# - 'Imbalanced' — legs don't sum to expected_net (L1 break)
# - 'Orphaned' — balanced but a Required child missing OR an XOR
# group has 0 or > 1 fired members (L2 chain break)
#
# Mirrors chain-instances completion_status semantics so the analyst
# sees consistent language across the Chains and Transfer Templates
# sheets.
TT_INSTANCES_CONTRACT = DatasetContract(columns=[
ColumnSpec("template_name", "STRING"),
ColumnSpec("transfer_id", "STRING"),
ColumnSpec("posting", "DATETIME"),
ColumnSpec("expected_net", "DECIMAL"),
ColumnSpec("actual_net", "DECIMAL"),
ColumnSpec("net_diff", "DECIMAL"),
ColumnSpec("leg_count", "INTEGER"),
ColumnSpec("completion_status", "STRING"),
])
# Per-leg row backing the Transfer Templates sheet's Sankey (M.3.10f).
# One row per leg of any current_transactions row carrying a
# template_name (i.e., legs that joined a TransferTemplate's shared
# Transfer). ``flow_source`` / ``flow_target`` derive from
# ``amount_direction`` so the Sankey reads as:
#
# debit account → template_name → credit account
# │
# ├──> matched chain child rail (M.3.10i)
# └──> orphan chain child rail (M.3.10i,
# synthetic row for declared edges
# the runtime didn't fire)
#
# Width = ABS(amount_money). Each leg contributes one segment to one
# side of the template middle-node. The shared template middle-node
# means a 4-leg shared Transfer renders as 2 source nodes + the
# template + 2 target nodes — natural multi-leg flow visualization.
#
# M.3.10i adds chain-child edges as additional flow segments coming
# OUT of the template node. ``edge_kind`` = 'template_leg' for the
# original legs; 'chain_matched' / 'chain_orphan' for the L2-declared
# chain children flowing from the template. Synthetic orphan rows
# are emitted per (parent firing, declared chain child) where no
# matched child exists, so the Sankey shows the FULL declared topology
# even when runtime data is incomplete.
#
# Shares ``template_name`` + ``posting`` + ``completion_status``
# columns with tt-instances so cross_dataset='ALL_DATASETS' filter
# groups apply BOTH the date + template + completion dropdowns to
# both datasets in lockstep — picking 'Complete' on the Completion
# dropdown narrows the Sankey and the Table together to just the
# matching firings (M.3.10k).
#
# Parameterized on pKey + pValues for the metadata cascade.
TT_LEGS_CONTRACT = DatasetContract(columns=[
ColumnSpec("template_name", "STRING"),
ColumnSpec("transfer_id", "STRING"),
ColumnSpec("posting", "DATETIME"),
ColumnSpec("account_name", "STRING"),
ColumnSpec("account_role", "STRING"),
ColumnSpec("amount_money", "DECIMAL"),
ColumnSpec("amount_direction", "STRING"),
ColumnSpec("amount_abs", "DECIMAL"),
ColumnSpec("flow_source", "STRING"),
ColumnSpec("flow_target", "STRING"),
ColumnSpec("edge_kind", "STRING"),
ColumnSpec("completion_status", "STRING"),
])
# -- L2 Exception contracts (M.3.7) ------------------------------------------
# L2.1 — required Chain entries where parent fired but child didn't.
# Subset of CHAINS_CONTRACT pre-filtered to (required + orphan_count > 0).
EXC_CHAIN_ORPHANS_CONTRACT = DatasetContract(columns=[
ColumnSpec("parent_name", "STRING"),
ColumnSpec("child_name", "STRING"),
ColumnSpec("parent_firing_count", "INTEGER"),
ColumnSpec("child_firing_count", "INTEGER"),
ColumnSpec("orphan_count", "INTEGER"),
])
# L2.2 — Posted Transactions whose rail_name doesn't match any
# declared Rail.rail_name.
EXC_UNMATCHED_RAIL_NAME_CONTRACT = DatasetContract(columns=[
ColumnSpec("rail_name", "STRING"),
ColumnSpec("posting_count", "INTEGER"),
])
# L2.3 — Rails declared in L2 with zero postings in the window.
EXC_DEAD_RAILS_CONTRACT = DatasetContract(columns=[
ColumnSpec("rail_name", "STRING"),
ColumnSpec("leg_shape", "STRING"),
])
# L2.4 — Aggregating-rail bundles_activity targets with zero matching
# activity in the window. Bundles_activity refs are Identifiers that
# the SPEC says match a Rail.name — the SQL checks the rail_name
# attribution.
EXC_DEAD_BUNDLES_ACTIVITY_CONTRACT = DatasetContract(columns=[
ColumnSpec("aggregating_rail", "STRING"),
ColumnSpec("bundle_target", "STRING"),
])
# L2.5 — Rail.metadata_keys declared in L2 that no posting carries a
# non-null value for in the window. Each row is one (rail_name,
# metadata_key) pair the L2 declared but the runtime never populated.
EXC_DEAD_METADATA_CONTRACT = DatasetContract(columns=[
ColumnSpec("rail_name", "STRING"),
ColumnSpec("metadata_key", "STRING"),
])
# L2.6 — LimitSchedule (parent_role, rail_name) cells with zero
# outbound debit flow in the window. Means the cap is effectively dead
# — either nobody routes that role/type combination, or the L2 declared
# a cap nobody enforces against.
EXC_DEAD_LIMIT_SCHEDULES_CONTRACT = DatasetContract(columns=[
ColumnSpec("parent_role", "STRING"),
ColumnSpec("rail_name", "STRING"),
ColumnSpec("cap", "DECIMAL"),
])
# Unified L2 Exceptions (M.3.10l) — UNION ALL across all 6 L2 hygiene
# checks with a shared shape so a single KPI + bar chart + detail
# table can present the whole L2-hygiene picture in one place.
# Mirrors the L1 dashboard's `_l1_exceptions` pattern (one row =
# one violation; check_type is the discriminator).
#
# - check_type: which L2 hygiene check produced the row.
# - entity_a / entity_b: the primary and secondary subject of the
# violation (e.g., parent rail + child rail for Chain Orphans;
# rail_name + metadata_key for Dead Metadata; rail_name alone
# for Unmatched Transfer Type with entity_b NULL).
# - detail: optional extra context (leg_shape, cap, etc.) — STRING
# regardless of source type so the unified projection works.
# - count: "how many violation rows of this kind", used for the bar
# chart's stacking + the detail table's sort order. Per check:
# * Chain Orphans → orphan_count (parent firings without a child)
# * Unmatched Transfer Type → posting_count (count of leaking legs)
# * Dead Rails / Dead Bundles / Dead Metadata / Dead Limit
# Schedules → 1 (each row IS one dead declaration)
# (AA.D.1: renamed from `magnitude`, which falsely implied a
# continuous-magnitude measure and was being rendered with
# currency=True. These are integer counts, not money amounts.)
UNIFIED_L2_EXCEPTIONS_CONTRACT = DatasetContract(columns=[
ColumnSpec("check_type", "STRING"),
# entity_a holds the L2-declared name relevant to each row's
# check_type — rail/template name for 4 of 6 checks, rail_name
# for L2.2, parent_role for L2.6. The shape lets the L2 Exceptions
# table's right-click drills wire entity_a → Rails sheet / Chains
# sheet filter parameters; the destination filters return zero
# rows for the 2 check_types whose entity_a isn't actually a rail
# or template name (transparent "this drill doesn't apply" UX).
ColumnSpec("entity_a", "STRING", shape=ColumnShape.L2_DECLARED_NAME),
ColumnSpec("entity_b", "STRING"),
ColumnSpec("detail", "STRING"),
# AO.6 — display as "Occurrences" so this per-violation magnitude
# column doesn't read as "Count" and collide with the sheet's
# "Open L2 Violations" KPI (which tallies violation ROWS). A reader
# otherwise sees "39 violations" above a column headed "Count" whose
# values run to the hundreds — the v11.9.4 cold-sweep's #6
# "count overloaded on L2 Exceptions".
# BH.11 rename (2026-05-25) — "Occurrences" was an earlier
# rename of bare "Count" but still left the unit-shift vs the
# KPI's row-count opaque to operators (cold-read finding #11:
# "KPI=41 / table column values in the thousands — two
# different units, same page, no signposting"). "Violations per
# Type" makes the per-row unit explicit against the KPI's
# "Distinct Exception Types Open" headline rename.
ColumnSpec("count", "INTEGER", display_name="Violations per Type"),
])
# -- Rails tab (M.3.10c) — postings explorer + cascade source ---------------
# Per-leg view from <prefix>_current_transactions, parameterized on
# pKey + pValues so the metadata cascade filter applies via QS
# CustomSql substitution. The Rails sheet's transactions Table reads
# directly from this dataset.
POSTINGS_CONTRACT = DatasetContract(columns=[
ColumnSpec("id", "STRING"),
ColumnSpec("transfer_id", "STRING"),
ColumnSpec("transfer_parent_id", "STRING"),
# rail_name is a drill destination for the L2 Exceptions table's
# "View in Rails" right-click — see UNIFIED_L2_EXCEPTIONS_CONTRACT.
ColumnSpec("rail_name", "STRING", shape=ColumnShape.L2_DECLARED_NAME),
ColumnSpec("account_id", "STRING"),
ColumnSpec("account_name", "STRING"),
ColumnSpec("account_role", "STRING"),
ColumnSpec("account_scope", "STRING"),
ColumnSpec("posting", "DATETIME"),
ColumnSpec("amount_money", "DECIMAL"),
ColumnSpec("amount_direction", "STRING"),
ColumnSpec("status", "STRING"),
ColumnSpec("bundle_id", "STRING"),
ColumnSpec("bundle_status", "STRING"), # 'Bundled' / 'Unbundled' calc
ColumnSpec("origin", "STRING"),
])
# Long-form (metadata_key, metadata_value) for the cascade. QS's
# CascadingControlConfiguration uses the metadata_key column to
# filter rows by the Key dropdown's selection — picking 'customer_id'
# in Key narrows the dataset to rows WHERE metadata_key='customer_id',
# then DISTINCT metadata_value populates the Value dropdown.
# (Earlier single-column shape with dataset-parameter substitution
# DIDN'T work — QS's cascade is a column-match filter, not a
# parameter-driven re-query.)
META_VALUES_CONTRACT = DatasetContract(columns=[
ColumnSpec("metadata_key", "STRING"),
ColumnSpec("metadata_value", "STRING"),
])
# -- Builders ----------------------------------------------------------------
[docs]
def build_all_l2_flow_tracing_datasets(
cfg: Config, l2_instance: L2Instance,
) -> list[DataSet]:
"""Return every Dataset the L2 Flow Tracing app needs.
Mirrors `build_all_l1_dashboard_datasets`: derives an L2-aware
``cfg`` (so dataset IDs carry the L2 instance prefix as their
middle segment per M.2d.3) when the caller hasn't pre-stamped it.
Idempotent — re-deriving an already-L2-aware cfg is a no-op.
M.3.6 ships Chains; M.3.7 adds the 6 L2 Exceptions sections;
M.3.10c adds the postings explorer + meta-values cascade source
for the Rails tab (replacing M.3.5's declared-rails table — moves
to a future Docs tab — and M.3.8's 28 per-key metadata dropdowns
— replaced by the cascade); M.3.10d swaps the chains aggregate
dataset for a per-parent-firing explorer (chain-instances);
M.3.10f adds the Transfer Templates sheet with tt-instances (per
shared Transfer) + tt-legs (per leg, backing the multi-leg flow
Sankey); M.3.10l replaces the 6 separate L2 exception datasets
with one unified UNION-ALL dataset (mirrors L1's exceptions
pattern — single KPI + bar chart + detail table).
"""
return [
build_postings_dataset(cfg, l2_instance),
build_meta_values_dataset(cfg, l2_instance),
build_chain_instances_dataset(cfg, l2_instance),
build_tt_instances_dataset(cfg, l2_instance),
build_tt_legs_dataset(cfg, l2_instance),
build_unified_l2_exceptions_dataset(cfg, l2_instance),
# M.4.4.5 — App Info ("i") sheet datasets, ALWAYS LAST.
# M.4.4.7 — per-app segment so deploy <single-app> doesn't
# delete-then-create another app's App Info dataset.
build_liveness_dataset(cfg, app_segment="l2ft"),
build_matview_status_dataset(
cfg, app_segment="l2ft",
view_specs=l2ft_matview_specs(cfg),
),
]
[docs]
def declared_rail_names(l2_instance: L2Instance) -> list[str]:
"""Sorted list of declared Rail names. Drives the Rail dropdown's
selectable values on the L2FT Rails sheet (X.1.b).
Pre-X.1.b the Rail dropdown carried no selectable_values and
relied on QS's auto-distinct fetch (``tenK-sample-values-V2``
endpoint), which 404s on cold per-CI-run dashboards. Static
enumeration from the L2 sidesteps the lazy fetch entirely —
rail names are bounded + known at deploy time.
"""
return sorted(str(r.name) for r in l2_instance.rails)
# X.1.b — bounded enums for the L2FT Rails sheet's Status + Bundle
# dropdowns. Hardcoded because:
# - ``status`` values are part of the L1 schema's ``CHECK``-style
# constraint (only ``Pending`` / ``Posted`` / ``Failed`` are valid).
# - ``bundle_status`` is a calc field defined as
# ``CASE WHEN bundle_id IS NULL THEN 'Unbundled' ELSE 'Bundled' END``
# — exactly two values, ever.
# StaticValues sourcing eliminates the QS auto-distinct fetch path
# (the X.1.b ``tenK-sample-values-V2`` 404 source).
# X.1.i — `status` is open-set in the L1 schema (any string), but only
# `Pending` / `Posted` carry first-class meaning in this tool (drives
# Aging, Conservation, Completion checks). Every other raw status
# (Failed, Cancelled, Rejected, ...) projects to `Other` via a CASE in
# the L2FT postings dataset SQL so this static enum matches what the
# column actually produces and the dropdown e2e (which asserts every
# advertised value has rows) doesn't choke on a stale enum.
_TRANSACTION_STATUS_VALUES: tuple[str, ...] = ("Pending", "Posted", "Other")
_BUNDLE_STATUS_VALUES: tuple[str, ...] = ("Bundled", "Unbundled")
# X.1.g — chain + TT completion_status enums. Mirror the CASE branches
# in build_chain_instances_dataset / build_tt_instances_dataset so QS
# dropdown options match the projected column values exactly.
_CHAIN_COMPLETION_STATUS_VALUES: tuple[str, ...] = (
"Completed", "Incomplete",
)
_TT_COMPLETION_STATUS_VALUES: tuple[str, ...] = (
"Complete", "Imbalanced", "Orphaned",
)
[docs]
def transaction_status_values() -> list[str]:
"""Bounded enum of transaction ``status`` values. Static dropdown
source — see ``_TRANSACTION_STATUS_VALUES`` for rationale.
"""
return list(_TRANSACTION_STATUS_VALUES)
[docs]
def bundle_status_values() -> list[str]:
"""Bounded enum of ``bundle_status`` calc-field values. Static
dropdown source — see ``_BUNDLE_STATUS_VALUES`` for rationale.
"""
return list(_BUNDLE_STATUS_VALUES)
[docs]
def chain_completion_status_values() -> list[str]:
"""Bounded enum of chain-instances ``completion_status`` values.
Static dropdown source on the Chains sheet (X.1.g).
"""
return list(_CHAIN_COMPLETION_STATUS_VALUES)
[docs]
def tt_completion_status_values() -> list[str]:
"""Bounded enum of transfer-template ``completion_status`` values.
Static dropdown source on the Transfer Templates sheet (X.1.g).
"""
return list(_TT_COMPLETION_STATUS_VALUES)
[docs]
def declared_chain_parents(l2_instance: L2Instance) -> list[str]:
"""Sorted list of distinct Chain row parent names. Drives the
Chain dropdown's selectable values on the Chains sheet (M.3.10d).
"""
return sorted({str(c.parent) for c in l2_instance.chains})
[docs]
def declared_template_names(l2_instance: L2Instance) -> list[str]:
"""Sorted list of declared TransferTemplate names. Drives the
Template dropdown's selectable values on the Transfer Templates
sheet (M.3.10f).
"""
return sorted(str(t.name) for t in l2_instance.transfer_templates)
[docs]
def build_postings_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""One row per leg from ``<prefix>_current_transactions``,
parameterized so the Rails sheet's filters push down into SQL via
QS ``<<$param>>`` substitution.
Two parameter families, both server-side:
- **Metadata cascade** — ``pKey`` (single) + ``pValues`` (single).
``pKey = '__ALL__'`` short-circuits the metadata WHERE to "no
filter" (freshly-loaded dashboard renders every leg);
``pValues = '__placeholder__'`` matches no real value, so picking
a Key without a Value goes empty (UX hint to pick both). Stays in
the inner query's WHERE.
- **Category pushdown (Y.2.c + AA.A.3)** — ``pL2ftRail`` /
``pL2ftStatus`` / ``pL2ftBundle`` (all SINGLE_VALUED post-AA.A.3;
was MULTI_VALUED pre-flip). Defaults to the L2FT_ALL_SENTINEL so a
freshly-loaded dashboard matches every row via the sentinel
disjunct in ``_match_all_in_clause``. Bridged from the Rails
sheet's SINGLE_SELECT dropdowns; clearing a dropdown reverts to
the sentinel default. Pushed into the OUTER WHERE because
``status`` and ``bundle_status`` are CASE-aliases (not visible to
a WHERE in the same SELECT) so the projection wraps in a subquery.
Phase BM — date narrowing pushes down via ``<<$pL2ftDateStart>>``
/ ``<<$pL2ftDateEnd>>`` (replaces the pre-BM per-sheet
``TimeRangeFilter`` on QS + no-narrow on App2). One SQL form
across both renderers.
"""
prefix = cfg.db_table_prefix
# AO.1.impl — amount_money is BIGINT cents on the base table; wrap
# to dollars at projection (inner subquery so the outer SELECT *
# passes the already-dollar form through).
amount = cents_to_dollars_sql("amount_money", dialect=cfg.dialect)
date_clause = _l2ft_date_range_clause(
"posting", P_L2FT_RAILS_DATE_START, P_L2FT_RAILS_DATE_END, cfg,
)
sql = (
f"SELECT * FROM (\n"
f" SELECT\n"
f" id, transfer_id, transfer_parent_id, rail_name,\n"
f" account_id, account_name, account_role, account_scope,\n"
f" posting, {amount} AS amount_money, amount_direction,\n"
# X.1.i — collapse open-set `status` into the bounded set the
# tool reasons about. Pending / Posted carry first-class meaning
# (drives Aging, Conservation, Completion checks); every other
# raw status (Failed, Cancelled, Rejected, ...) projects to
# 'Other' so the static dropdown enum matches what the column
# produces and the analyst can still narrow to the unhealthy
# tail without enumerating every possible terminal state.
f" CASE WHEN status IN ('Pending', 'Posted') THEN status "
f"ELSE 'Other' END AS status,\n"
f" bundle_id,\n"
f" CASE WHEN bundle_id IS NULL THEN 'Unbundled' ELSE 'Bundled' END "
f"AS bundle_status,\n"
f" origin\n"
f" FROM {prefix}_current_transactions\n"
f" WHERE\n"
# The metadata cascade short-circuit: when pKey is the sentinel,
# this sub-clause always evaluates true (no filtering); otherwise
# the per-key branches compare the leg's metadata against the
# picked values. See `metadata_filter_clause` for the
# per-dialect-safe WHERE shape.
f"{metadata_filter_clause(l2_instance, 'metadata', cfg.dialect)}"
f" AND {date_clause}\n"
f") postings\n"
# AA.A.3 — rail / status / bundle all pushed into SQL via the
# single-valued sentinel-guard form (`('sentinel' = <<$p>> OR col
# = <<$p>>)`). Pre-AA.A.3 status / bundle (fixed ≤3-element enums)
# had a multi-valued value-list default; AA.A.3 flipped every
# pushdown dropdown to SINGLE per the drill-to-one default.
# Clearing a dropdown reverts to the sentinel default (= all rows).
f"WHERE {_match_all_in_clause('rail_name', 'pL2ftRail')}\n"
f" AND {_match_all_in_clause('status', 'pL2ftStatus')}\n"
f" AND {_match_all_in_clause('bundle_status', 'pL2ftBundle')}\n"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-postings-dataset"),
"L2FT Postings", "l2ft-postings",
sql, POSTINGS_CONTRACT,
visual_identifier=DS_POSTINGS,
dataset_parameters=[
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pKey",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[META_KEY_ALL_SENTINEL],
),
)),
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pValues",
# Y.1.m: SINGLE_VALUED to match the analysis-level
# parameter shape (text-field control). Was MULTI_VALUED
# but the text-field control couldn't commit non-empty
# values to multi-valued params — broke the cascade.
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[META_VALUE_PLACEHOLDER_SENTINEL],
),
)),
# AA.A.3 — rail / status / bundle SINGLE_VALUED pushdown.
# All three default to the L2FT_ALL_SENTINEL (the SQL
# `_match_all_in_clause` makes the sentinel mean "all" on
# load). Pre-AA.A.3 status / bundle had value-list defaults
# and rail_name had the sentinel-1-element-list default
# (X.2.t.2). AA.A.3 unified all three on the SINGLE shape.
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pL2ftRail",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[L2FT_ALL_SENTINEL],
),
)),
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pL2ftStatus",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[L2FT_ALL_SENTINEL],
),
)),
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pL2ftBundle",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[L2FT_ALL_SENTINEL],
),
)),
*_l2ft_match_all_range_params(
P_L2FT_RAILS_DATE_START, P_L2FT_RAILS_DATE_END,
),
],
)
[docs]
def build_chains_dataset(cfg: Config, l2_instance: L2Instance) -> DataSet:
"""One row per declared Chain-row child (Z.A: a multi-children
row contributes N rows) — the L2's parent→child topology joined
to runtime parent firing counts + matched-child counts.
A row IS one Sankey edge in the Chains visual. Counts come from
``<prefix>_current_transactions`` matched on the parent's name
(which can be a Rail's ``rail_name`` OR a TransferTemplate's
``template_name`` — every leg row carries both, with template_name
taking precedence when a rail is part of a template). Child
matches require ``transfer_parent_id`` to point at one of the
parent's transfer_ids — that's the runtime "did this child fire
in response to this parent" relation.
Orphan rate = (parent_firings without a matched child) /
parent_firings. A required Chain entry with non-zero orphan rate
is the seed for M.3.7's L2.1 'Chain orphans' exception.
Note on portability: uses correlated subqueries instead of
ARRAY_AGG (PG-only) to keep the SQL portable. The chains table is
small (typically tens of entries), so the cost is bounded.
"""
prefix = cfg.db_table_prefix
declared = _declared_chains_cte(cfg)
sql = (
f"WITH declared AS (\n{declared}\n),\n"
f"edge_runtime AS (\n"
f" SELECT\n"
f" d.parent_name,\n"
f" d.child_name,\n"
f" d.required,\n"
f" d.xor_group,\n"
f" d.source_node,\n"
f" d.target_node,\n"
f" COALESCE((\n"
f" SELECT COUNT(DISTINCT t.transfer_id)\n"
f" FROM {prefix}_current_transactions t\n"
f" WHERE COALESCE(t.template_name, t.rail_name) = d.parent_name\n"
f" ), 0) AS parent_firing_count,\n"
f" COALESCE((\n"
f" SELECT COUNT(DISTINCT c.transfer_id)\n"
f" FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) = d.child_name\n"
f" AND c.transfer_parent_id IN (\n"
f" SELECT t2.transfer_id\n"
f" FROM {prefix}_current_transactions t2\n"
f" WHERE COALESCE(t2.template_name, t2.rail_name) "
f"= d.parent_name\n"
f" )\n"
f" ), 0) AS child_firing_count\n"
f" FROM declared d\n"
f")\n"
f"SELECT\n"
f" e.parent_name,\n"
f" e.child_name,\n"
f" e.required,\n"
f" e.xor_group,\n"
f" e.source_node,\n"
f" e.target_node,\n"
f" e.parent_firing_count,\n"
f" e.child_firing_count,\n"
# GREATEST clamps at 0 — child can fire more than parent in some
# patterns (e.g., one parent triggers many children); negative
# orphans don't read intuitively in the visual.
f" {greatest('e.parent_firing_count - e.child_firing_count', '0', dialect=cfg.dialect)} "
f"AS orphan_count,\n"
f" CASE\n"
f" WHEN e.parent_firing_count > 0\n"
f" THEN CAST(\n"
f" {greatest('e.parent_firing_count - e.child_firing_count', '0', dialect=cfg.dialect)} "
f"AS DECIMAL(20,4)\n"
f" ) / e.parent_firing_count\n"
f" ELSE 0\n"
f" END AS orphan_rate\n"
f"FROM edge_runtime e\n"
f"ORDER BY e.parent_name, e.child_name"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-chains-dataset"),
"L2FT Chains", "l2ft-chains",
sql, CHAINS_CONTRACT,
visual_identifier=DS_CHAINS,
)
[docs]
def build_chain_instances_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""One row per parent transfer firing of a declared chain parent
(M.3.10d, completion_status extended in M.3.10i for XOR groups).
Backs the Chains sheet's per-instance explorer.
Columns:
- ``parent_chain_name`` — the L2-declared parent rail / template
name. Drives the Chain dropdown's selectable values.
- ``parent_transfer_id`` — DISTINCT transfer_id of the parent
firing. Multiple legs of one transfer collapse to one row via
GROUP BY.
- ``completion_status`` — computed inline; one of:
* ``'Completed'`` — every Required child fired AND every XOR
group has exactly one member fired.
* ``'Incomplete'`` — at least one Required child missing OR
any XOR group has 0 fired (orphan) OR > 1 fired (violation).
The pre-X.1.j third branch ``'No Required Children'`` is gone —
validator rule C5 rejects all-optional / no-XOR chains at L2
load, so the SQL never produces that case.
- ``parent_metadata`` is read in the WHERE only — kept off the
contract so users don't see raw JSON. ``pKey`` / ``pValues``
substitute into a JSONPath ``IN (...)`` predicate same as the
postings dataset; the ``__ALL__`` sentinel short-circuits to
"no metadata filter".
SQL portability: correlated subqueries (no ``ARRAY_AGG``); no
JSONB; ``MAX`` aggregates over varchar status / metadata which
isn't perfect but the parent transfer's legs share these values
in practice. The chains table is bounded by L2 declarations
(typically tens of entries) so the cost stays predictable.
"""
prefix = cfg.db_table_prefix
# AO.1.impl — parent_amount_money flows from MAX(t.amount_money)
# in BIGINT cents through two CTEs unchanged; wrap at the outermost
# projection so the chain-instances dataset surfaces dollars.
parent_amt = cents_to_dollars_sql(
"parent_amount_money", dialect=cfg.dialect,
)
declared = _declared_chains_cte(cfg)
sql = (
f"WITH declared AS (\n{declared}\n),\n"
f"parent_chains AS (\n"
f" SELECT\n"
f" parent_name,\n"
f" SUM(CASE WHEN required = 'Required' THEN 1 ELSE 0 END) "
f"AS required_total,\n"
f" COUNT(DISTINCT CASE WHEN xor_group IS NOT NULL "
f"THEN xor_group END) AS xor_group_count\n"
f" FROM declared\n"
f" GROUP BY parent_name\n"
f"),\n"
f"parent_firings AS (\n"
f" SELECT\n"
f" pc.parent_name AS parent_chain_name,\n"
f" pc.required_total,\n"
f" pc.xor_group_count,\n"
f" t.transfer_id AS parent_transfer_id,\n"
f" MIN(t.posting) AS parent_posting,\n"
f" MAX(t.status) AS parent_status,\n"
f" MAX(t.amount_money) AS parent_amount_money,\n"
f" MAX(t.metadata) AS parent_metadata\n"
f" FROM parent_chains pc\n"
f" JOIN {prefix}_current_transactions t\n"
f" ON COALESCE(t.template_name, t.rail_name) = pc.parent_name\n"
f" GROUP BY pc.parent_name, pc.required_total, pc.xor_group_count, "
f"t.transfer_id\n"
f"),\n"
f"firing_completion AS (\n"
f" SELECT\n"
f" pf.parent_chain_name,\n"
f" pf.parent_transfer_id,\n"
f" pf.parent_posting,\n"
f" pf.parent_status,\n"
f" pf.parent_amount_money,\n"
f" pf.required_total,\n"
f" pf.xor_group_count,\n"
f" pf.parent_metadata,\n"
f" (\n"
f" SELECT COUNT(DISTINCT d.child_name)\n"
f" FROM declared d\n"
f" WHERE d.parent_name = pf.parent_chain_name\n"
f" AND d.required = 'Required'\n"
f" AND EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) "
f"= d.child_name\n"
f" AND c.transfer_parent_id = pf.parent_transfer_id\n"
f" )\n"
f" ) AS required_fired,\n"
# XOR violation count = number of declared XOR groups under this
# parent where fired-children count != 1. SPEC: "exactly one of
# them SHOULD fire per parent instance"; 0 fired = orphan, > 1
# fired = violation. Both flagged as Incomplete.
f" (\n"
f" SELECT COUNT(*)\n"
f" FROM (\n"
f" SELECT d.xor_group,\n"
f" SUM(CASE WHEN EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) "
f"= d.child_name\n"
f" AND c.transfer_parent_id = pf.parent_transfer_id\n"
f" ) THEN 1 ELSE 0 END) AS fired_in_group\n"
f" FROM declared d\n"
f" WHERE d.parent_name = pf.parent_chain_name\n"
f" AND d.xor_group IS NOT NULL\n"
f" GROUP BY d.xor_group\n"
f" ) g\n"
f" WHERE g.fired_in_group <> 1\n"
f" ) AS xor_violations\n"
f" FROM parent_firings pf\n"
f")\n"
# Y.2.d — wrap the projection in a subquery so the CASE-aliased
# `completion_status` is visible to the outer WHERE; `parent_chain_name`
# joins it there. Metadata cascade on `parent_metadata` stays inner
# (the column isn't projected, so the WHERE that reads it must be).
f"SELECT * FROM (\n"
f" SELECT\n"
f" parent_chain_name,\n"
f" parent_transfer_id,\n"
f" parent_posting,\n"
f" parent_status,\n"
f" {parent_amt} AS parent_amount_money,\n"
f" required_total,\n"
f" required_fired,\n"
f" CASE\n"
f" WHEN required_fired >= required_total "
f"AND xor_violations = 0 THEN 'Completed'\n"
f" ELSE 'Incomplete'\n"
f" END AS completion_status\n"
f" FROM firing_completion\n"
f" WHERE\n"
f"{metadata_filter_clause(l2_instance, 'parent_metadata', cfg.dialect)}\n"
f") chain_instances\n"
# AA.A.3 — chain / completion SINGLE_VALUED pushdown via the
# sentinel-guard form. Pre-AA.A.3 completion had a value-list
# default; AA.A.3 flipped every pushdown to SINGLE per the
# drill-to-one default. Clearing a dropdown reverts to the
# sentinel default (= all rows match).
# Phase BM — Chains-sheet date pickers narrow ``parent_posting``
# via the unified ``<<$pL2ftChainsDate*>>`` pushdown (replaces
# the pre-BM TimeRangeFilter FG).
f"WHERE {_match_all_in_clause('parent_chain_name', 'pL2ftChainsChain')}\n"
f" AND {_match_all_in_clause('completion_status', 'pL2ftChainsCompletion')}\n"
f" AND {_l2ft_date_range_clause('parent_posting', P_L2FT_CHAINS_DATE_START, P_L2FT_CHAINS_DATE_END, cfg)}\n"
f"ORDER BY parent_posting DESC"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-chain-instances-dataset"),
"L2FT Chain Instances", "l2ft-chain-instances",
sql, CHAIN_INSTANCES_CONTRACT,
visual_identifier=DS_CHAIN_INSTANCES,
dataset_parameters=[
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pKey",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[META_KEY_ALL_SENTINEL],
),
)),
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pValues",
# Y.1.m: SINGLE_VALUED to match the analysis-level
# parameter shape (text-field control). Was MULTI_VALUED
# but the text-field control couldn't commit non-empty
# values to multi-valued params — broke the cascade.
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[META_VALUE_PLACEHOLDER_SENTINEL],
),
)),
# AA.A.3 — chain / completion SINGLE_VALUED pushdown. Both
# default to the L2FT_ALL_SENTINEL (`_match_all_in_clause`
# makes the sentinel mean "all" on load). An instance with
# no Chains declared lands on the same sentinel default —
# harmless, the chain-instances table is empty anyway.
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pL2ftChainsChain",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[L2FT_ALL_SENTINEL],
),
)),
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pL2ftChainsCompletion",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[L2FT_ALL_SENTINEL],
),
)),
*_l2ft_match_all_range_params(
P_L2FT_CHAINS_DATE_START, P_L2FT_CHAINS_DATE_END,
),
],
)
# -- L2 Exception builders (M.3.7) -------------------------------------------
[docs]
def build_exc_chain_orphans_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""L2.1 — Required Chain entries where parent fired but no
matched child fired in the window.
Reuses the chains dataset's CTE shape (declared edges + edge
runtime) and filters to ``required = 'Required' AND orphan_count
> 0``. XOR-group multi/none violations are deferred to a follow-on
substep — a precise XOR check needs per-Transfer-id grouping that
the simpler aggregate doesn't capture.
AJ.4 (Gap I): a ``fan_in`` (N:1) chain is singleton-children, so it's
labeled ``Required`` and would take the naive
``parent_firing_count − child_firing_count`` path — but N parents
converging on far fewer shared child Transfers makes that difference
large and positive on *healthy* batching, so every fan_in chain read
as a pile of orphans. For fan_in edges the orphan count is instead
the number of parent firings whose ``transfer_id`` is NOT a parent of
any child batch Transfer (the genuine "cycle closed but never
assigned to a batch"), derived from the AB.4.3 ``_transfer_parents``
matview. This is production-correct: fan_in is a real structural
property of the L2, not a demo artifact.
"""
prefix = cfg.db_table_prefix
declared = _declared_chains_cte(cfg)
sql = (
f"WITH declared AS (\n{declared}\n),\n"
f"edge_runtime AS (\n"
f" SELECT\n"
f" d.parent_name,\n"
f" d.child_name,\n"
f" d.required,\n"
f" d.fan_in,\n"
f" COALESCE((\n"
f" SELECT COUNT(DISTINCT t.transfer_id)\n"
f" FROM {prefix}_current_transactions t\n"
f" WHERE COALESCE(t.template_name, t.rail_name) = d.parent_name\n"
f" ), 0) AS parent_firing_count,\n"
f" COALESCE((\n"
f" SELECT COUNT(DISTINCT c.transfer_id)\n"
f" FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) = d.child_name\n"
f" AND c.transfer_parent_id IN (\n"
f" SELECT t2.transfer_id\n"
f" FROM {prefix}_current_transactions t2\n"
f" WHERE COALESCE(t2.template_name, t2.rail_name) "
f"= d.parent_name\n"
f" )\n"
f" ), 0) AS child_firing_count,\n"
# AJ.4 (Gap I): fan_in-only — parent firings not assigned to any
# batch. CASE-guarded so the _transfer_parents anti-join only runs
# for fan_in edges (0 for the naive non-fan_in path).
f" CASE WHEN d.fan_in = 1 THEN COALESCE((\n"
f" SELECT COUNT(DISTINCT t3.transfer_id)\n"
f" FROM {prefix}_current_transactions t3\n"
f" WHERE COALESCE(t3.template_name, t3.rail_name) "
f"= d.parent_name\n"
f" AND t3.transfer_id NOT IN (\n"
f" SELECT tp.parent_transfer_id\n"
f" FROM {prefix}_transfer_parents tp\n"
f" JOIN {prefix}_current_transactions ch\n"
f" ON ch.transfer_id = tp.child_transfer_id\n"
f" WHERE COALESCE(ch.template_name, ch.rail_name) "
f"= d.child_name\n"
f" )\n"
f" ), 0) ELSE 0 END AS unbatched_parent_count\n"
f" FROM declared d\n"
f"),\n"
f"edge_orphans AS (\n"
f" SELECT\n"
f" e.parent_name,\n"
f" e.child_name,\n"
f" e.required,\n"
f" e.parent_firing_count,\n"
f" e.child_firing_count,\n"
f" CASE WHEN e.fan_in = 1 THEN e.unbatched_parent_count\n"
f" ELSE {greatest('e.parent_firing_count - e.child_firing_count', '0', dialect=cfg.dialect)} "
f"END AS orphan_count\n"
f" FROM edge_runtime e\n"
f")\n"
f"SELECT\n"
f" e.parent_name,\n"
f" e.child_name,\n"
f" e.parent_firing_count,\n"
f" e.child_firing_count,\n"
f" e.orphan_count\n"
f"FROM edge_orphans e\n"
f"WHERE e.required = 'Required'\n"
f" AND e.orphan_count > 0\n"
f"ORDER BY e.orphan_count DESC, e.parent_name, e.child_name"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-exc-chain-orphans-dataset"),
"L2 Exc — Chain Orphans", "l2ft-exc-chain-orphans",
sql, EXC_CHAIN_ORPHANS_CONTRACT,
visual_identifier=DS_EXC_CHAIN_ORPHANS,
)
[docs]
def build_exc_unmatched_rail_name_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""L2.2 — Posted Transactions whose ``rail_name`` doesn't
match any declared ``Rail.rail_name``.
The runtime version of M.2d.1's deferred validator check
('every Transfer MUST match a Rail'). LEFT JOIN to a CTE of
declared types + filter to NULL surfaces the unmatched rows.
Output is per-transfer-type with a count of postings carrying
that type — the table reveals what's leaking past the L2's rails.
"""
prefix = cfg.db_table_prefix
declared = _declared_rail_names_cte(cfg)
sql = (
f"WITH declared_types AS (\n{declared}\n)\n"
f"SELECT\n"
f" t.rail_name,\n"
f" COUNT(*) AS posting_count\n"
f"FROM {prefix}_current_transactions t\n"
f"LEFT JOIN declared_types d ON d.rail_name = t.rail_name\n"
f"WHERE d.rail_name IS NULL\n"
f"GROUP BY t.rail_name\n"
f"ORDER BY posting_count DESC, t.rail_name"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-exc-unmatched-rail-name-dataset"),
"L2 Exc — Unmatched Rail Name",
"l2ft-exc-unmatched-rail-name",
sql, EXC_UNMATCHED_RAIL_NAME_CONTRACT,
visual_identifier=DS_EXC_UNMATCHED_RAIL_NAME,
)
[docs]
def build_exc_dead_rails_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""L2.3 — Rails declared in L2 with zero postings in the window.
Same shape as the Rails dataset but pre-filtered to
``COALESCE(r.total_postings, 0) = 0``. The KPI shows the count;
the detail table lists the dead rails so the integrator can
decide whether to retire the declaration or fix the ETL.
"""
prefix = cfg.db_table_prefix
declared = _declared_rails_cte(l2_instance, cfg.dialect)
sql = (
f"WITH declared AS (\n{declared}\n),\n"
f"runtime AS (\n"
f" SELECT rail_name, COUNT(*) AS total_postings\n"
f" FROM {prefix}_current_transactions\n"
f" GROUP BY rail_name\n"
f")\n"
f"SELECT\n"
f" d.rail_name,\n"
f" d.leg_shape\n"
f"FROM declared d\n"
f"LEFT JOIN runtime r ON r.rail_name = d.rail_name\n"
f"WHERE COALESCE(r.total_postings, 0) = 0\n"
f"ORDER BY d.rail_name"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-exc-dead-rails-dataset"),
"L2 Exc — Dead Rails", "l2ft-exc-dead-rails",
sql, EXC_DEAD_RAILS_CONTRACT,
visual_identifier=DS_EXC_DEAD_RAILS,
)
[docs]
def build_exc_dead_bundles_activity_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""L2.4 — Aggregating-rail bundles_activity targets that no
posting matched in the window.
Per Z.B (2026-05-15): bundles_activity refs are Identifier rail
names; the SQL checks ``t.rail_name = db.bundle_target``. Each row
is one (aggregating_rail, target) pair the L2 declared but the
runtime never realized.
"""
prefix = cfg.db_table_prefix
declared = _declared_bundles_activity_cte(l2_instance, cfg.dialect)
sql = (
f"WITH declared_bundles AS (\n{declared}\n)\n"
f"SELECT\n"
f" db.aggregating_rail,\n"
f" db.bundle_target\n"
f"FROM declared_bundles db\n"
f"WHERE NOT EXISTS (\n"
f" SELECT 1\n"
f" FROM {prefix}_current_transactions t\n"
f" WHERE t.rail_name = db.bundle_target\n"
f")\n"
f"ORDER BY db.aggregating_rail, db.bundle_target"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-exc-dead-bundles-activity-dataset"),
"L2 Exc — Dead Bundles Activity",
"l2ft-exc-dead-bundles-activity",
sql, EXC_DEAD_BUNDLES_ACTIVITY_CONTRACT,
visual_identifier=DS_EXC_DEAD_BUNDLES_ACTIVITY,
)
[docs]
def build_exc_dead_limit_schedules_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""L2.6 — LimitSchedule (parent_role, rail_name) cells with
zero outbound debit flow in the window.
Means the cap is effectively dead — either nobody routes that
role/type combination, or the L2 declared a cap nobody enforces
against. NOT EXISTS over the prefixed transactions matview keeps
the query bounded by the (small) limit-schedule count.
"""
prefix = cfg.db_table_prefix
declared = _declared_limit_schedules_cte(cfg)
sql = (
f"WITH declared_limits AS (\n{declared}\n)\n"
f"SELECT\n"
f" dl.parent_role,\n"
f" dl.rail_name,\n"
f" dl.cap\n"
f"FROM declared_limits dl\n"
f"WHERE NOT EXISTS (\n"
f" SELECT 1\n"
f" FROM {prefix}_current_transactions t\n"
f" WHERE t.account_parent_role = dl.parent_role\n"
f" AND t.rail_name = dl.rail_name\n"
f" AND t.amount_direction = 'Debit'\n"
f")\n"
f"ORDER BY dl.parent_role, dl.rail_name"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-exc-dead-limit-schedules-dataset"),
"L2 Exc — Dead Limit Schedules",
"l2ft-exc-dead-limit-schedules",
sql, EXC_DEAD_LIMIT_SCHEDULES_CONTRACT,
visual_identifier=DS_EXC_DEAD_LIMIT_SCHEDULES,
)
# -- Unified L2 Exceptions (M.3.10l) -----------------------------------------
[docs]
def build_unified_l2_exceptions_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""UNION ALL across the 6 L2 hygiene checks into one row-per-
violation dataset (M.3.10l).
Mirrors L1's `l1_exceptions` pattern: one row = one violation;
`check_type` is the discriminator; `count` is the per-violation row
count used for sort + bar stacking. Each branch inlines its own
CTEs as a subquery so the outer SELECT can do consistent typing
across branches without colliding CTE names.
Each branch is functionally equivalent to one of the 6 retired
`build_exc_*` queries, just projected to the unified shape via
CASTs + literal `check_type` labels.
"""
prefix = cfg.db_table_prefix
declared_chains = _declared_chains_cte(cfg)
declared_types = _declared_rail_names_cte(cfg)
declared_rails = _declared_rails_cte(l2_instance, cfg.dialect)
declared_bundles = _declared_bundles_activity_cte(l2_instance, cfg.dialect)
declared_limits = _declared_limit_schedules_cte(cfg)
dead_metadata_fragments = _dead_metadata_check_fragments(
l2_instance, prefix, cfg.dialect,
)
if dead_metadata_fragments:
dead_metadata_inner = "\n UNION ALL\n".join(dead_metadata_fragments)
else:
nt = typed_null("varchar(4000)", cfg.dialect)
df = dual_from(cfg.dialect)
dead_metadata_inner = (
f" SELECT {nt} AS rail_name, "
f"{nt} AS metadata_key{df} WHERE 1=0"
)
sql = (
# Branch 1: Chain Orphans
f"SELECT\n"
f" CAST('Chain Orphans' AS VARCHAR(50)) AS check_type,\n"
f" parent_name AS entity_a,\n"
f" child_name AS entity_b,\n"
f" CAST(NULL AS VARCHAR(255)) AS detail,\n"
f" CAST(orphan_count AS INTEGER) AS count\n"
f"FROM (\n"
f" WITH declared AS (\n{declared_chains}\n),\n"
f" edge_runtime AS (\n"
f" SELECT\n"
f" d.parent_name, d.child_name, d.required,\n"
f" COALESCE((\n"
f" SELECT COUNT(DISTINCT t.transfer_id)\n"
f" FROM {prefix}_current_transactions t\n"
f" WHERE COALESCE(t.template_name, t.rail_name) "
f"= d.parent_name\n"
f" ), 0) AS parent_firing_count,\n"
f" COALESCE((\n"
f" SELECT COUNT(DISTINCT c.transfer_id)\n"
f" FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) "
f"= d.child_name\n"
f" AND c.transfer_parent_id IN (\n"
f" SELECT t2.transfer_id\n"
f" FROM {prefix}_current_transactions t2\n"
f" WHERE COALESCE(t2.template_name, t2.rail_name) "
f"= d.parent_name\n"
f" )\n"
f" ), 0) AS child_firing_count\n"
f" FROM declared d\n"
f" )\n"
f" SELECT parent_name, child_name,\n"
f" {greatest('parent_firing_count - child_firing_count', '0', dialect=cfg.dialect)} "
f"AS orphan_count\n"
f" FROM edge_runtime\n"
f" WHERE required = 'Required'\n"
f" AND parent_firing_count > child_firing_count\n"
f") sub_chain_orphans\n"
# Branch 2: Unmatched Rail Name (Z.B renamed from
# "Unmatched Transfer Type"; the check itself was always
# rail-keyed — only the label was stale).
f"UNION ALL\n"
f"SELECT\n"
f" CAST('Unmatched Rail Name' AS VARCHAR(50)),\n"
f" CAST(rail_name AS VARCHAR(255)),\n"
f" CAST(NULL AS VARCHAR(255)),\n"
f" CAST(NULL AS VARCHAR(255)),\n"
f" CAST(posting_count AS INTEGER)\n"
f"FROM (\n"
f" WITH declared_types AS (\n{declared_types}\n)\n"
f" SELECT t.rail_name, COUNT(*) AS posting_count\n"
f" FROM {prefix}_current_transactions t\n"
f" LEFT JOIN declared_types d "
f"ON d.rail_name = t.rail_name\n"
f" WHERE d.rail_name IS NULL\n"
f" GROUP BY t.rail_name\n"
f") sub_unmatched\n"
# Branch 3: Dead Rails
f"UNION ALL\n"
f"SELECT\n"
f" CAST('Dead Rails' AS VARCHAR(50)),\n"
f" CAST(rail_name AS VARCHAR(255)),\n"
f" CAST(NULL AS VARCHAR(255)),\n"
f" CAST(leg_shape AS VARCHAR(255)),\n"
f" 1\n"
f"FROM (\n"
f" WITH declared AS (\n{declared_rails}\n),\n"
f" runtime AS (\n"
f" SELECT rail_name, COUNT(*) AS total_postings\n"
f" FROM {prefix}_current_transactions GROUP BY rail_name\n"
f" )\n"
f" SELECT d.rail_name, d.leg_shape\n"
f" FROM declared d\n"
f" LEFT JOIN runtime r ON r.rail_name = d.rail_name\n"
f" WHERE COALESCE(r.total_postings, 0) = 0\n"
f") sub_dead_rails\n"
# Branch 4: Dead Bundles Activity
f"UNION ALL\n"
f"SELECT\n"
f" CAST('Dead Bundles Activity' AS VARCHAR(50)),\n"
f" CAST(aggregating_rail AS VARCHAR(255)),\n"
f" CAST(bundle_target AS VARCHAR(255)),\n"
f" CAST(NULL AS VARCHAR(255)),\n"
f" 1\n"
f"FROM (\n"
f" WITH declared_bundles AS (\n{declared_bundles}\n)\n"
f" SELECT db.aggregating_rail, db.bundle_target\n"
f" FROM declared_bundles db\n"
f" WHERE NOT EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions t\n"
f" WHERE t.rail_name = db.bundle_target "
f"OR t.rail_name = db.bundle_target\n"
f" )\n"
f") sub_dead_bundles\n"
# Branch 5: Dead Metadata Declarations
f"UNION ALL\n"
f"SELECT\n"
f" CAST('Dead Metadata Declarations' AS VARCHAR(50)),\n"
f" CAST(rail_name AS VARCHAR(255)),\n"
f" CAST(metadata_key AS VARCHAR(255)),\n"
f" CAST(NULL AS VARCHAR(255)),\n"
f" 1\n"
f"FROM (\n{dead_metadata_inner}\n) sub_dead_metadata\n"
# Branch 6: Dead Limit Schedules
f"UNION ALL\n"
f"SELECT\n"
f" CAST('Dead Limit Schedules' AS VARCHAR(50)),\n"
f" CAST(parent_role AS VARCHAR(255)),\n"
f" CAST(rail_name AS VARCHAR(255)),\n"
f" CAST(cap AS VARCHAR(255)),\n"
f" 1\n"
f"FROM (\n"
f" WITH declared_limits AS (\n{declared_limits}\n)\n"
f" SELECT dl.parent_role, dl.rail_name, dl.cap\n"
f" FROM declared_limits dl\n"
f" WHERE NOT EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions t\n"
f" WHERE t.account_parent_role = dl.parent_role\n"
f" AND t.rail_name = dl.rail_name\n"
f" AND t.amount_direction = 'Debit'\n"
f" )\n"
f") sub_dead_limits\n"
# Position-based — Oracle's ORDER BY after UNION ALL doesn't
# recognize aliases when each branch carries WITH+CTE subqueries.
# Columns: 1=check_type, 2=entity_a, 3=entity_b, 4=detail, 5=count.
f"ORDER BY 5 DESC, 1, 2, 3"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-unified-exceptions-dataset"),
"L2 Unified Exceptions", "l2ft-unified-exceptions",
sql, UNIFIED_L2_EXCEPTIONS_CONTRACT,
visual_identifier=DS_UNIFIED_L2_EXCEPTIONS,
)
# -- Internals ---------------------------------------------------------------
def _declared_rails_cte(l2_instance: L2Instance, dialect: Dialect) -> str:
"""Render the L2-declared rails as a UNION ALL of SELECT-literal rows.
UNION ALL of single-row SELECTs is used instead of ``VALUES (...)``
so each column gets a CAST (or naked literal) that's resolved per
row, avoiding the "type of column N is text but row M is null"
inference problem PostgreSQL's planner sometimes hits with VALUES
when most rows have NULL for a column.
``dialect`` selects the typed-NULL form for the empty-rails fallback
branch (``NULL::TEXT`` on Postgres, ``CAST(NULL AS CLOB)`` on
Oracle).
"""
df = dual_from(dialect)
if not l2_instance.rails:
# Should not happen for a valid L2 (there must be some rails);
# the validator would catch it. Return a safe empty CTE that
# produces zero rows so the LEFT JOIN works.
nt = typed_null("varchar(4000)", dialect)
return (
" SELECT\n"
f" {nt} AS rail_name,\n"
f" {nt} AS leg_shape,\n"
f" {nt} AS source_role,\n"
f" {nt} AS destination_role,\n"
f" {nt} AS leg_role,\n"
f" {nt} AS max_pending_age,\n"
f" {nt} AS max_unbundled_age,\n"
f" {nt} AS posted_requirements"
f"{df}\n"
" WHERE 1=0"
)
rows: list[str] = []
for r in l2_instance.rails:
leg_shape = _leg_shape(r)
if isinstance(r, TwoLegRail):
source_role = _role_str(r.source_role)
destination_role = _role_str(r.destination_role)
leg_role = None
else:
source_role = None
destination_role = None
leg_role = _role_str(r.leg_role)
max_pending = _duration_label(r.max_pending_age)
max_unbundled = _duration_label(r.max_unbundled_age)
posted_reqs = ",".join(
sorted(str(k) for k in posted_requirements_for(l2_instance, r.name))
)
rows.append(
" SELECT "
f"{_sql_str(str(r.name))} AS rail_name, "
f"{_sql_str(leg_shape)} AS leg_shape, "
f"{_sql_nullable_str(source_role)} AS source_role, "
f"{_sql_nullable_str(destination_role)} AS destination_role, "
f"{_sql_nullable_str(leg_role)} AS leg_role, "
f"{_sql_nullable_str(max_pending)} AS max_pending_age, "
f"{_sql_nullable_str(max_unbundled)} AS max_unbundled_age, "
f"{_sql_str(posted_reqs)} AS posted_requirements"
f"{df}"
)
return "\n UNION ALL\n".join(rows)
def _declared_chains_cte(cfg: Config) -> str:
"""Declared-chain-children CTE sourced from the BS.5 projection view.
Reads ``<prefix>_v_config_chain_children`` (BS.5 — itself a relational
walk over ``<prefix>_config_kv``) and decorates each row with the
Required/Optional + xor_group + source_node/target_node columns the
chain consumers expect.
Column semantics (preserved from the pre-BS.5 UNION-ALL-of-literals
shape):
- ``required`` — ``'Required'`` when this child is the only sibling
under its chain parent (Z.A: singleton ⇒ required); ``'Optional'``
otherwise.
- ``xor_group`` — a stable per-group key (NULL for singletons,
``parent_name`` itself for multi-children groups; consumers only
use it as a GROUP BY / DISTINCT key, not parsed for the literal
shape).
- ``source_node`` / ``target_node`` — Sankey display strings (still
identical to parent / child name; the seam exists so M.3.6+ can
decorate without breaking JOIN semantics).
- ``fan_in`` — INTEGER 0/1 (AJ.4 — chain_orphans uses this to bypass
naive parent-minus-child subtraction for N:1 children).
Static-collapse: emit no longer enumerates L2 chains in Python; the
same SQL ships regardless of how many chains the L2 declares. The
view's GROUP BY does the projection.
"""
p = cfg.db_table_prefix
return (
f" SELECT\n"
f" parent_name,\n"
f" child_name,\n"
f" CASE WHEN sibling_count = 1 THEN 'Required'\n"
f" ELSE 'Optional' END AS required,\n"
f" CASE WHEN sibling_count > 1 THEN parent_name END AS xor_group,\n"
f" parent_name AS source_node,\n"
f" child_name AS target_node,\n"
f" fan_in\n"
f" FROM (\n"
f" SELECT parent_name, child_name, fan_in,\n"
f" COUNT(*) OVER (PARTITION BY parent_name) AS sibling_count\n"
f" FROM {p}_v_config_chain_children\n"
f" ) chain_children_with_count"
)
def _leg_shape(rail: TwoLegRail | SingleLegRail) -> str:
"""Compact label combining leg arity + aggregating flag.
Examples: "1-leg" / "2-leg" / "1-leg-aggregating" / "2-leg-aggregating".
"""
arity = "2-leg" if isinstance(rail, TwoLegRail) else "1-leg"
return f"{arity}-aggregating" if rail.aggregating else arity
def _role_str(role: RoleExpression) -> str:
"""Render a RoleExpression (tuple of Identifiers) as a display string.
Single-role: the name. UNION (multi-role): joined with " | " (the
SPEC's union-set notation)."""
parts = [str(r) for r in role]
if not parts:
return ""
if len(parts) == 1:
return parts[0]
return " | ".join(parts)
def _duration_label(td: timedelta | None) -> str | None:
"""Render a timedelta as a compact label ("24h", "1d", "30m").
None → None (the SQL emits NULL). Non-evenly-divisible durations
fall back to seconds.
"""
if td is None:
return None
s = int(td.total_seconds())
if s == 0:
return "0s"
if s % 86400 == 0:
return f"{s // 86400}d"
if s % 3600 == 0:
return f"{s // 3600}h"
if s % 60 == 0:
return f"{s // 60}m"
return f"{s}s"
def _sql_str(value: str) -> str:
"""Escape a Python string for embedding as a SQL string literal.
Doubles single quotes per SQL standard (works on PostgreSQL +
portable to other RDBMS per the project's portability constraint)."""
return "'" + value.replace("'", "''") + "'"
def _sql_nullable_str(value: str | None) -> str:
"""SQL literal for an optional string — emits NULL when None."""
if value is None:
return "NULL"
return _sql_str(value)
# -- M.3.7 CTE helpers -------------------------------------------------------
def _declared_rail_names_cte(cfg: Config) -> str:
"""Distinct rail-type identifiers sourced from the AW projection view.
BS.5 (2026-05-29): swapped a UNION-ALL-of-literal-rows for a SELECT
against ``<prefix>_v_config_rails``. Each rail's ``name`` IS the
type identifier (Z.B 2026-05-15 symmetric collapse). The view's
GROUP BY emits one row per L2 Rail; this CTE re-projects the
``name`` column under the legacy ``rail_name`` alias (Z.B.12
deferred — ``transactions.rail_name`` is still the column).
"""
p = cfg.db_table_prefix
return f" SELECT name AS rail_name FROM {p}_v_config_rails"
def _declared_bundles_activity_cte(
l2_instance: L2Instance, dialect: Dialect,
) -> str:
"""All (aggregating_rail, bundle_target) pairs the L2 declares.
bundle_target is whatever Identifier the rail's
``bundles_activity`` lists — per SPEC, that resolves to either a
rail_name or a rail_name at runtime.
"""
pairs: list[tuple[str, str]] = []
for r in l2_instance.rails:
if not r.aggregating:
continue
for target in r.bundles_activity:
pairs.append((str(r.name), str(target)))
df = dual_from(dialect)
if not pairs:
nt = typed_null("varchar(4000)", dialect)
return (
f" SELECT {nt} AS aggregating_rail, "
f"{nt} AS bundle_target{df} WHERE 1=0"
)
rows = [
f" SELECT {_sql_str(agg)} AS aggregating_rail, "
f"{_sql_str(target)} AS bundle_target{df}"
for agg, target in pairs
]
return "\n UNION ALL\n".join(rows)
def _dead_metadata_check_fragments(
l2_instance: L2Instance, prefix: str, dialect: Dialect,
) -> list[str]:
"""One SELECT per declared (rail, metadata_key) pair guarded by
NOT EXISTS against the prefixed transactions matview.
Static JSON paths (``$.<literal>``) keep the SQL portable —
PostgreSQL doesn't accept dynamic JSONPath arguments to
``JSON_VALUE`` without the v17+ JSON_TABLE syntax, and the
project's no-JSONB constraint rules out the ``->>`` shortcut.
"""
df = dual_from(dialect)
fragments: list[str] = []
for r in l2_instance.rails:
for key in r.metadata_keys:
rail_name = str(r.name)
key_name = str(key)
json_path = f"$.{key_name}"
jv = json_value("t.metadata", _sql_str(json_path), dialect)
fragments.append(
f" SELECT {_sql_str(rail_name)} AS rail_name, "
f"{_sql_str(key_name)} AS metadata_key"
f"{df}\n"
f" WHERE NOT EXISTS (\n"
f" SELECT 1\n"
f" FROM {prefix}_current_transactions t\n"
f" WHERE t.rail_name = {_sql_str(rail_name)}\n"
f" AND t.metadata IS NOT NULL\n"
f" AND {jv} IS NOT NULL\n"
f" )"
)
return fragments
def _declared_limit_schedules_cte(cfg: Config) -> str:
"""Per-LimitSchedule rows sourced from the AW projection view.
BS.5 (2026-05-29): swapped a UNION-ALL-of-literal-rows for a SELECT
against ``<prefix>_v_config_limit_schedules``. The view projects
one row per L2 LimitSchedule with typed ``parent_role`` / ``rail``
/ ``direction`` / ``cap``; this CTE re-aliases ``rail`` to
``rail_name`` (Z.B.12 deferred) and downcasts ``cap`` to
``DECIMAL(20,2)`` for the consumers' aggregation shape.
"""
p = cfg.db_table_prefix
return (
f" SELECT parent_role,\n"
f" rail AS rail_name,\n"
f" CAST(cap AS DECIMAL(20,2)) AS cap\n"
f" FROM {p}_v_config_limit_schedules"
)
# -- Transfer Templates sheet (M.3.10f) ------------------------------------
def _declared_templates_cte(
l2_instance: L2Instance, dialect: Dialect,
) -> str:
"""Render L2-declared TransferTemplate names + expected_net as a
UNION ALL of SELECT-literal rows. The tt-instances builder joins
against this CTE so only declared templates appear in the dataset
(any rogue ``template_name`` value in current_transactions that
doesn't correspond to a declared TransferTemplate is excluded —
surfaced separately by the L2.2 unmatched-transfer-type check).
AO.1.impl — emits ``expected_net`` in BIGINT cents so the
callers' internal cents-vs-cents math against SUM(amount_money)
(also cents post-foundation) compares like units. Outer
projections wrap the result back to dollars via
``cents_to_dollars_sql``. The L2-authored cap stays a Decimal in
dollars; multiplying by 100 in this emitter is the cleanest way
to lift it into the cents number space without round-trip float
error (the *100 is an integer multiplication on Python's
Decimal, then ``CAST AS DECIMAL(20,0)`` keeps the SQL literal
integer-shaped).
"""
df = dual_from(dialect)
if not l2_instance.transfer_templates:
return (
f" SELECT {typed_null('varchar(4000)', dialect)} AS template_name, "
f"{typed_null('numeric', dialect)} AS expected_net"
f"{df} WHERE 1=0"
)
rows: list[str] = []
for t in l2_instance.transfer_templates:
expected_cents = int(t.expected_net * 100)
rows.append(
f" SELECT {_sql_str(str(t.name))} AS template_name, "
f"CAST({expected_cents} AS DECIMAL(20,0)) AS expected_net{df}"
)
return "\n UNION ALL\n".join(rows)
[docs]
def build_tt_instances_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""One row per shared Transfer that matches a declared
TransferTemplate (M.3.10f, completion_status reshaped M.3.10j).
A "shared Transfer" is one ``transfer_id`` from
``<prefix>_current_transactions`` whose legs all carry the same
``template_name`` matching a declared template. Per SPEC: every
firing of a ``leg_rails`` rail with the same ``transfer_key``
Metadata values posts to the same shared Transfer, so the
transfer_id distinct-count = number of TransferTemplate
instances.
``completion_status`` is one of:
- 'Imbalanced' — ``ABS(actual_net - expected_net) >= 0.01``
(L1 Conservation break).
- 'Orphaned' — balanced, but a Required chain child didn't fire
OR an XOR group has 0 or > 1 fired members (L2 chain break).
- 'Complete' — balanced AND every Required child fired AND every
XOR group has exactly one fired member.
Mirrors the chain-instances completion_status semantics so the
analyst sees consistent language across both sheets.
Parameterized on pKey + pValues for the metadata cascade.
"""
prefix = cfg.db_table_prefix
# AO.1.impl — internal CTE math runs in BIGINT cents (expected_net
# was multiplied ×100 in the declared-templates CTE; actual_net is
# SUM(amount_money) which is cents from the foundation). The
# cents-vs-cents subtraction for ``net_diff`` is integer-safe; wrap
# all three money columns back to dollars at the outer projection.
actual = cents_to_dollars_sql("actual_net", dialect=cfg.dialect)
expected = cents_to_dollars_sql("expected_net", dialect=cfg.dialect)
netdiff = cents_to_dollars_sql(
"(actual_net - expected_net)", dialect=cfg.dialect,
)
declared_tt = _declared_templates_cte(l2_instance, cfg.dialect)
declared_ch = _declared_chains_cte(cfg)
sql = (
f"WITH templates AS (\n{declared_tt}\n),\n"
f"declared AS (\n{declared_ch}\n),\n"
# Chain-shape per template: counts of declared Required children
# + count of distinct XOR groups. Templates with no chain entries
# (parent_name not in declared) get NULL for the LEFT JOIN, which
# the COALESCEs below treat as zero.
f"template_chain_shape AS (\n"
f" SELECT\n"
f" t.template_name,\n"
f" COALESCE(SUM(CASE WHEN d.required = 'Required' "
f"THEN 1 ELSE 0 END), 0) AS required_total,\n"
f" COUNT(DISTINCT CASE WHEN d.xor_group IS NOT NULL "
f"THEN d.xor_group END) AS xor_group_count\n"
f" FROM templates t\n"
f" LEFT JOIN declared d ON d.parent_name = t.template_name\n"
f" GROUP BY t.template_name\n"
f"),\n"
f"firings AS (\n"
f" SELECT\n"
f" t.template_name,\n"
f" t.expected_net,\n"
f" tcs.required_total,\n"
f" tcs.xor_group_count,\n"
f" ct.transfer_id,\n"
f" MIN(ct.posting) AS posting,\n"
f" SUM(ct.amount_money) AS actual_net,\n"
f" COUNT(*) AS leg_count,\n"
f" MAX(ct.metadata) AS parent_metadata\n"
f" FROM templates t\n"
f" JOIN template_chain_shape tcs ON tcs.template_name = t.template_name\n"
f" JOIN {prefix}_current_transactions ct\n"
f" ON ct.template_name = t.template_name\n"
f" GROUP BY t.template_name, t.expected_net, tcs.required_total, "
f"tcs.xor_group_count, ct.transfer_id\n"
f"),\n"
# Chain completeness per firing — same shape as chain-instances:
# required_fired = how many declared-Required children were
# matched via transfer_parent_id; xor_violations = how many
# declared XOR groups had ≠ 1 fired members.
f"firing_completion AS (\n"
f" SELECT\n"
f" f.*,\n"
f" (\n"
f" SELECT COUNT(DISTINCT d.child_name)\n"
f" FROM declared d\n"
f" WHERE d.parent_name = f.template_name\n"
f" AND d.required = 'Required'\n"
f" AND EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) "
f"= d.child_name\n"
f" AND c.transfer_parent_id = f.transfer_id\n"
f" )\n"
f" ) AS required_fired,\n"
f" (\n"
f" SELECT COUNT(*)\n"
f" FROM (\n"
f" SELECT d.xor_group,\n"
f" SUM(CASE WHEN EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) "
f"= d.child_name\n"
f" AND c.transfer_parent_id = f.transfer_id\n"
f" ) THEN 1 ELSE 0 END) AS fired_in_group\n"
f" FROM declared d\n"
f" WHERE d.parent_name = f.template_name\n"
f" AND d.xor_group IS NOT NULL\n"
f" GROUP BY d.xor_group\n"
f" ) g\n"
f" WHERE g.fired_in_group <> 1\n"
f" ) AS xor_violations\n"
f" FROM firings f\n"
f")\n"
# Y.2.e — wrap the projection so the CASE-aliased
# `completion_status` is visible to the outer WHERE; `template_name`
# joins it there. Metadata cascade on `parent_metadata` stays inner.
f"SELECT * FROM (\n"
f" SELECT\n"
f" template_name,\n"
f" transfer_id,\n"
f" posting,\n"
f" {expected} AS expected_net,\n"
f" {actual} AS actual_net,\n"
f" {netdiff} AS net_diff,\n"
f" leg_count,\n"
f" CASE\n"
f" WHEN ABS(actual_net - expected_net) >= 1 THEN 'Imbalanced'\n"
f" WHEN required_fired < required_total THEN 'Orphaned'\n"
f" WHEN xor_violations > 0 THEN 'Orphaned'\n"
f" ELSE 'Complete'\n"
f" END AS completion_status\n"
f" FROM firing_completion\n"
f" WHERE\n"
f"{metadata_filter_clause(l2_instance, 'parent_metadata', cfg.dialect)}\n"
f") tt_instances\n"
# AA.A.3 — template / completion SINGLE_VALUED pushdown via the
# sentinel-guard form. Pre-AA.A.3 completion had a value-list
# default; AA.A.3 flipped every pushdown to SINGLE per the
# drill-to-one default. Clearing a dropdown reverts to the
# sentinel default (= all rows match).
# Phase BM — TT-sheet date pickers narrow ``posting`` via
# ``<<$pL2ftTtDate*>>`` (shared with tt_legs so both Sankey +
# Table narrow together — the pre-BM ALL_DATASETS TimeRangeFilter
# FG dissolves).
f"WHERE {_match_all_in_clause('template_name', 'pL2ftTtTemplate')}\n"
f" AND {_match_all_in_clause('completion_status', 'pL2ftTtCompletion')}\n"
f" AND {_l2ft_date_range_clause('posting', P_L2FT_TT_DATE_START, P_L2FT_TT_DATE_END, cfg)}\n"
f"ORDER BY posting DESC, template_name, transfer_id"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-tt-instances-dataset"),
"L2FT TT Instances", "l2ft-tt-instances",
sql, TT_INSTANCES_CONTRACT,
visual_identifier=DS_TT_INSTANCES,
dataset_parameters=_tt_dataset_parameters(l2_instance),
)
def _tt_dataset_parameters(
l2_instance: L2Instance,
) -> list[DatasetParameter]:
"""Y.2.e + AA.A.3 — the shared dataset-parameter set for both
TransferTemplates datasets (tt-instances + tt-legs): the metadata
cascade pair (pKey / pValues) plus the template / completion pushdown
pair (both SINGLE_VALUED post-AA.A.3). AK.1 — build_dataset assigns
the Ids deterministically (auto_id over dataset_id + param name), so
the two datasets get distinct Ids for the same param name.
"""
return [
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pKey",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[META_KEY_ALL_SENTINEL],
),
)),
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pValues",
# Y.1.m: SINGLE_VALUED to match the analysis-level
# parameter shape (text-field control). Was MULTI_VALUED
# but the text-field control couldn't commit non-empty
# values to multi-valued params — broke the cascade.
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[META_VALUE_PLACEHOLDER_SENTINEL],
),
)),
# AA.A.3 — template / completion SINGLE_VALUED pushdown pair.
# Both default to the L2FT_ALL_SENTINEL (`_match_all_in_clause`
# makes the sentinel mean "all" on load). An instance with no
# Templates declared lands on the same sentinel default —
# harmless, the tt-instances / tt-legs tables are empty anyway.
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pL2ftTtTemplate",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[L2FT_ALL_SENTINEL],
),
)),
DatasetParameter(StringDatasetParameter=StringDatasetParameter(
Name="pL2ftTtCompletion",
ValueType="SINGLE_VALUED",
DefaultValues=StringDatasetParameterDefaultValues(
StaticValues=[L2FT_ALL_SENTINEL],
),
)),
*_l2ft_match_all_range_params(
P_L2FT_TT_DATE_START, P_L2FT_TT_DATE_END,
),
]
[docs]
def build_tt_legs_dataset(
cfg: Config, l2_instance: L2Instance,
) -> DataSet:
"""One row per Sankey edge segment for a TransferTemplate firing
(M.3.10f, chain-edge UNION added in M.3.10i).
The query UNIONs three row-sources:
1. **Template legs** (``edge_kind='template_leg'``) — actual
transactions where ``template_name`` matches a declared
TransferTemplate. ``flow_source`` / ``flow_target`` derive from
``amount_direction`` so the Sankey reads as
``debit account → template_name → credit account``.
2. **Matched chain children** (``edge_kind='chain_matched'``) —
actual transactions whose ``transfer_parent_id`` points at a
template firing's transfer_id AND whose rail_name matches a
declared chain child. ``flow_source = template_name``,
``flow_target = child_rail_name``.
3. **Orphan chain children** (``edge_kind='chain_orphan'``) —
SYNTHETIC rows for declared (template, child) chain edges
that the runtime didn't fire. One row per (parent firing,
declared child) where no matching child transaction exists.
``flow_source = template_name``, ``flow_target = child_rail_name
|| ' (orphan)'`` so the analyst sees the missing edge
visualized as a thin dashed-style ribbon to a distinct node.
Together this gives a complete Sankey per parent firing — every
declared chain edge is visible whether it fired or not, with the
template legs forming the central trunk.
Width = ABS(amount_money). Synthetic orphan rows carry the parent
firing's amount as a representative ribbon thickness so they're
visually present but recognizable.
Joining against the declared-templates CTE filters out any
rogue ``template_name`` value in current_transactions that isn't
in the L2 declaration (mirrors tt-instances).
Parameterized on pKey + pValues for the metadata cascade.
"""
prefix = cfg.db_table_prefix
# AO.1.impl — wrap money columns at the outermost UNION projection.
# Inner CTEs run in BIGINT cents (amount_money from base table;
# expected_net pre-multiplied ×100 in the declared-templates CTE;
# `ABS(actual_net - expected_net) >= 1` is cents-vs-cents).
amount = cents_to_dollars_sql("amount_money", dialect=cfg.dialect)
amount_abs = cents_to_dollars_sql("amount_abs", dialect=cfg.dialect)
declared_tt = _declared_templates_cte(l2_instance, cfg.dialect)
declared_ch = _declared_chains_cte(cfg)
sql = (
f"WITH templates AS (\n{declared_tt}\n),\n"
f"declared AS (\n{declared_ch}\n),\n"
# Per-firing completion calc — same shape as tt-instances'
# firing_completion CTE so the column denormalized into every
# leg + chain-edge row matches what the Table sees per firing.
# Lets cross_dataset='ALL_DATASETS' on the Completion filter
# narrow both Sankey + Table together (M.3.10k).
f"firing_completion AS (\n"
f" SELECT\n"
f" t.template_name,\n"
f" t.expected_net,\n"
f" ct.transfer_id,\n"
f" SUM(ct.amount_money) AS actual_net,\n"
f" (\n"
f" SELECT COUNT(*)\n"
f" FROM declared d\n"
f" WHERE d.parent_name = t.template_name\n"
f" AND d.required = 'Required'\n"
f" ) AS required_total,\n"
f" (\n"
f" SELECT COUNT(DISTINCT d.child_name)\n"
f" FROM declared d\n"
f" WHERE d.parent_name = t.template_name\n"
f" AND d.required = 'Required'\n"
f" AND EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) "
f"= d.child_name\n"
f" AND c.transfer_parent_id = ct.transfer_id\n"
f" )\n"
f" ) AS required_fired,\n"
f" (\n"
f" SELECT COUNT(*) FROM (\n"
f" SELECT d.xor_group,\n"
f" SUM(CASE WHEN EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions c\n"
f" WHERE COALESCE(c.template_name, c.rail_name) "
f"= d.child_name\n"
f" AND c.transfer_parent_id = ct.transfer_id\n"
f" ) THEN 1 ELSE 0 END) AS fired_in_group\n"
f" FROM declared d\n"
f" WHERE d.parent_name = t.template_name\n"
f" AND d.xor_group IS NOT NULL\n"
f" GROUP BY d.xor_group\n"
f" ) g WHERE g.fired_in_group <> 1\n"
f" ) AS xor_violations\n"
f" FROM templates t\n"
f" JOIN {prefix}_current_transactions ct\n"
f" ON ct.template_name = t.template_name\n"
f" GROUP BY t.template_name, t.expected_net, ct.transfer_id\n"
f"),\n"
f"firing_status AS (\n"
f" SELECT\n"
f" transfer_id,\n"
f" CASE\n"
f" WHEN ABS(actual_net - expected_net) >= 1 THEN 'Imbalanced'\n"
f" WHEN required_fired < required_total THEN 'Orphaned'\n"
f" WHEN xor_violations > 0 THEN 'Orphaned'\n"
f" ELSE 'Complete'\n"
f" END AS completion_status\n"
f" FROM firing_completion\n"
f"),\n"
# Real template legs.
f"template_legs AS (\n"
f" SELECT\n"
f" ct.template_name,\n"
f" ct.transfer_id,\n"
f" ct.posting,\n"
f" ct.metadata,\n"
f" ct.account_name,\n"
f" ct.account_role,\n"
f" ct.amount_money,\n"
f" ct.amount_direction,\n"
f" ABS(ct.amount_money) AS amount_abs,\n"
f" CASE\n"
f" WHEN ct.amount_direction = 'Debit' THEN ct.account_name\n"
f" ELSE ct.template_name\n"
f" END AS flow_source,\n"
f" CASE\n"
f" WHEN ct.amount_direction = 'Debit' THEN ct.template_name\n"
f" ELSE ct.account_name\n"
f" END AS flow_target,\n"
f" CAST('template_leg' AS VARCHAR(20)) AS edge_kind,\n"
f" fs.completion_status\n"
f" FROM {prefix}_current_transactions ct\n"
f" JOIN templates t ON t.template_name = ct.template_name\n"
f" JOIN firing_status fs ON fs.transfer_id = ct.transfer_id\n"
f"),\n"
# One row per (parent firing, declared chain child) — the
# cartesian we need to detect both matched + orphan edges.
# parent_firings dedupes the legs of one shared Transfer to
# one (template_name, transfer_id, posting, metadata) row so
# the cartesian doesn't multiply by leg count.
f"parent_firings AS (\n"
f" SELECT DISTINCT\n"
f" template_name,\n"
f" transfer_id,\n"
f" MIN(posting) OVER (PARTITION BY transfer_id) AS posting,\n"
f" MAX(metadata) OVER (PARTITION BY transfer_id) AS metadata,\n"
f" MAX(ABS(amount_money)) OVER (PARTITION BY transfer_id) "
f"AS firing_amount_abs,\n"
f" MAX(completion_status) OVER (PARTITION BY transfer_id) "
f"AS completion_status\n"
f" FROM template_legs\n"
f"),\n"
f"chain_edges AS (\n"
f" SELECT\n"
f" pf.template_name,\n"
f" pf.transfer_id,\n"
f" pf.posting,\n"
f" pf.metadata,\n"
f" CAST(NULL AS VARCHAR(255)) AS account_name,\n"
f" CAST(NULL AS VARCHAR(100)) AS account_role,\n"
f" pf.firing_amount_abs AS amount_money,\n"
f" CAST('Credit' AS VARCHAR(20)) AS amount_direction,\n"
f" pf.firing_amount_abs AS amount_abs,\n"
f" pf.template_name AS flow_source,\n"
f" CASE WHEN EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions c\n"
f" WHERE c.rail_name = d.child_name\n"
f" AND c.transfer_parent_id = pf.transfer_id\n"
f" ) THEN d.child_name\n"
f" ELSE d.child_name || ' (orphan)'\n"
f" END AS flow_target,\n"
f" CASE WHEN EXISTS (\n"
f" SELECT 1 FROM {prefix}_current_transactions c\n"
f" WHERE c.rail_name = d.child_name\n"
f" AND c.transfer_parent_id = pf.transfer_id\n"
f" ) THEN CAST('chain_matched' AS VARCHAR(20))\n"
f" ELSE CAST('chain_orphan' AS VARCHAR(20))\n"
f" END AS edge_kind,\n"
f" pf.completion_status\n"
f" FROM parent_firings pf\n"
f" JOIN declared d ON d.parent_name = pf.template_name\n"
f")\n"
# Y.2.e — wrap the UNION so `template_name` / `completion_status`
# (both real columns in each branch's projection) are visible to a
# single outer WHERE. Metadata cascade on `metadata` stays inside
# each branch (the column is in the CTEs but not the projection).
f"SELECT * FROM (\n"
f" SELECT template_name, transfer_id, posting, account_name, "
f"account_role, "
f"{amount} AS amount_money, amount_direction, "
f"{amount_abs} AS amount_abs, "
f"flow_source, flow_target, edge_kind, completion_status\n"
f" FROM template_legs\n"
f" WHERE\n"
f"{metadata_filter_clause(l2_instance, 'metadata', cfg.dialect)}\n"
f" UNION ALL\n"
f" SELECT template_name, transfer_id, posting, account_name, "
f"account_role, "
f"{amount} AS amount_money, amount_direction, "
f"{amount_abs} AS amount_abs, "
f"flow_source, flow_target, edge_kind, completion_status\n"
f" FROM chain_edges\n"
f" WHERE\n"
f"{metadata_filter_clause(l2_instance, 'metadata', cfg.dialect)}\n"
f") tt_legs\n"
# AA.A.3 — template / completion SINGLE_VALUED pushdown (mirrors
# tt-instances so the Template / Completion dropdowns narrow the
# Sankey + Table together — the M.3.10k denormalization made this
# pair available).
# Phase BM — same TT-sheet date narrowing on ``posting`` as
# tt-instances; both datasets carry the BM-shape ``<<$pL2ftTtDate*>>``
# placeholders so the picker pair drives them in lockstep.
f"WHERE {_match_all_in_clause('template_name', 'pL2ftTtTemplate')}\n"
f" AND {_match_all_in_clause('completion_status', 'pL2ftTtCompletion')}\n"
f" AND {_l2ft_date_range_clause('posting', P_L2FT_TT_DATE_START, P_L2FT_TT_DATE_END, cfg)}\n"
f"ORDER BY posting DESC, template_name, transfer_id"
)
return build_dataset(
cfg, cfg.prefixed("l2ft-tt-legs-dataset"),
"L2FT TT Legs", "l2ft-tt-legs",
sql, TT_LEGS_CONTRACT,
visual_identifier=DS_TT_LEGS,
dataset_parameters=_tt_dataset_parameters(l2_instance),
)