"""Editor primitives — server-owned cascade for the X.4.e editor flow.
Three transforms on the in-memory ``L2Instance``:
- ``mutate_l2(instance, kind, id, fields)`` — replace one entity's
fields with the operator-supplied values, return a new ``L2Instance``.
Field-level only (no cross-entity ripple — that's rename's job).
- ``rename_identifier(instance, kind, old, new)`` — rewrite every
reference to ``old`` across the model. Symmetric to the strict
validator's reference-resolution pass: where the validator says
"this Rail's source_role MUST resolve to an Account.role", rename
rewrites those very fields when an Account.role changes.
- ``delete_l2_entity(instance, kind, id)`` — remove one entity + run
the validator. A structural break (some other entity still
referenced the deleted one) raises ``L2ValidationError``; the
caller (Studio's PUT handler) returns 400 with the validator
message inline.
All three return a new ``L2Instance`` (the ``L2InstanceCache.replace``
contract from X.4.a.6) — the original is never mutated. The cache +
disk-write pair handle persistence; this module is the pure-Python
transform layer.
Severability: pure Python; no DB, no Starlette. Imports the model +
validator only.
"""
from __future__ import annotations
import dataclasses
from collections.abc import Mapping
from typing import Any, Literal, TypeAlias, cast
from recon_gen.common.l2.primitives import (
Account,
AccountTemplate,
Chain,
ChainChildSpec,
Identifier,
L2Instance,
LimitSchedule,
Money,
Name,
Rail,
RailName,
SingleLegRail,
TransferTemplate,
TwoLegRail,
)
EntityKind: TypeAlias = Literal[
"account",
"account_template",
"rail",
"transfer_template",
"chain",
"limit_schedule",
# X.4.f.12 — singletons (one per L2 instance, not a list).
# Routes mount at ``/l2_shape/<kind>/`` (no entity_id); each
# renders a single edit form.
"theme",
# BXa.1 (2026-05-30): the ``persona`` singleton + its structured
# form went with the deleted ``DemoPersona`` dataclass. Institution
# name + acronym now live on the instance singleton's structured
# form alongside ``description``.
# AI.2.c — top-level L2Instance settings (``description`` + new
# post-BXa.1 fields ``institution_name`` / ``institution_acronym`` +
# ``role_business_day_offsets``) edited via a structured form
# (BXa.1 upgraded from raw YAML textarea).
"instance",
]
# X.4.f.12 — kinds that exist as a single optional attribute on
# L2Instance rather than a tuple. Routes / handlers branch on this
# to skip list view, create page, delete, and per-id addressing.
SINGLETON_KINDS: frozenset[EntityKind] = frozenset(
{"theme", "instance"},
)
# ---------------------------------------------------------------------------
# Public surface
# ---------------------------------------------------------------------------
[docs]
def mutate_l2(
instance: L2Instance,
kind: EntityKind,
entity_id: str,
fields: Mapping[str, Any], # typing-smell: ignore[explicit-any]: heterogeneous form-submitted field values; per-entity dataclass fields differ
) -> L2Instance:
"""Replace one entity's fields with new values.
Args:
instance: The L2 model to mutate (returns a new copy; original
untouched).
kind: Which collection the entity lives in.
entity_id: The entity's identity key — Account.id, Rail.name,
TransferTemplate.name, AccountTemplate.role,
Chain's "<parent>::<sorted-children-csv>" composite, or
LimitSchedule's "<parent_role>::<rail>" composite.
fields: New field values, applied via ``dataclasses.replace``.
Keys MUST match the dataclass field names; unknown keys
raise ``ValueError``.
Returns:
A new ``L2Instance`` with the matched entity replaced.
Raises:
KeyError: no entity with that ``entity_id`` exists in the
target collection.
ValueError: ``fields`` contains keys that aren't dataclass
fields of the target entity.
"""
matched, idx, collection = _find_entity(instance, kind, entity_id)
new_entity = dataclasses.replace(matched, **fields)
new_collection = collection[:idx] + (new_entity,) + collection[idx + 1:]
return _replace_collection(instance, kind, new_collection)
[docs]
def rename_identifier(
instance: L2Instance,
kind: EntityKind,
old: Identifier,
new: Identifier,
) -> L2Instance:
"""Rename an identifier across every L2 reference.
Per the SPEC's editor cascade rule: "Rename = auto-rewrite refs.
Renaming an identifier walks the model and replaces every field
that references the old value." The reference catalog mirrors the
strict validator's reference-resolution pass — wherever the
validator says "this field MUST resolve to ``old``", the rename
rewrites that field to ``new``.
Per kind:
- **account / account_template** (ID = role): walks every
``role`` / ``parent_role`` / ``source_role`` / ``destination_role``
/ ``leg_role`` field; rewrites RoleExpression tuples element-wise.
- **rail** (ID = name): rewrites ``leg_rails`` (TransferTemplate),
``bundles_activity`` (Rail), ``parent`` / ``children`` (Chain).
- **transfer_template** (ID = name): rewrites
``bundles_activity`` (Rail), ``parent`` / ``children`` (Chain).
- **chain / limit_schedule**: have no incoming references — rename
is a no-op (chains/limit_schedules are leaf consumers).
The Account.id / AccountTemplate (no .id, addressed by .role) /
LimitSchedule (composite key) are addressing keys, not reference
targets inside L2 — renaming Account.id walks the Account itself
only (rename via ``mutate_l2(..., fields={"id": new})``).
Returns a new ``L2Instance``; original untouched. Does NOT run
validation — caller composes ``validate(...)`` if cascade resulted
in an invalid model (e.g., renaming to a value that collides with
another entity's identifier).
"""
if kind in ("chain", "limit_schedule"):
return instance # no incoming refs to rewrite
if kind in ("account", "account_template"):
return _rename_role(instance, old, new)
if kind == "rail":
return _rename_rail(instance, old, new)
# kind == "transfer_template"
return _rename_transfer_template(instance, old, new)
[docs]
def attach_rail_to_reconciler(
instance: L2Instance,
*,
new_rail_name: str,
reconciler_kind: str,
reconciler_name: str,
) -> L2Instance:
"""BB.1 — append ``new_rail_name`` to an existing reconciler's
rail-list field so a freshly-created non-aggregating single-leg
rail satisfies S3 / C3 reconciliation atomically with its create.
``reconciler_kind``:
- ``"transfer_template"`` — append to ``TransferTemplate.leg_rails``
- ``"aggregating_rail"`` — append to ``Rail.bundles_activity``
(the rail must already be ``aggregating=True``)
Returns a new ``L2Instance`` with the reconciler's field updated.
Caller composes ``validate(...)`` afterward.
Raises:
KeyError: no reconciler with that ``reconciler_name`` exists.
ValueError: ``reconciler_kind`` not recognized, or the matched
reconciler isn't a valid receiver (e.g., a non-aggregating
rail picked as ``aggregating_rail``).
"""
name_id = Identifier(new_rail_name)
if reconciler_kind == "transfer_template":
for i, t in enumerate(instance.transfer_templates):
if str(t.name) == reconciler_name:
if name_id in t.leg_rails:
return instance # idempotent (already attached)
new_t = dataclasses.replace(
t, leg_rails=(*t.leg_rails, name_id),
)
new_tts = (
instance.transfer_templates[:i]
+ (new_t,)
+ instance.transfer_templates[i + 1:]
)
return dataclasses.replace(
instance, transfer_templates=new_tts,
)
raise KeyError(
f"TransferTemplate {reconciler_name!r} not found",
)
if reconciler_kind == "aggregating_rail":
for i, r in enumerate(instance.rails):
if str(r.name) != reconciler_name:
continue
if not r.aggregating:
raise ValueError(
f"Rail {reconciler_name!r} is not aggregating; "
f"can't be a reconciler"
)
if name_id in r.bundles_activity:
return instance # idempotent
new_r = dataclasses.replace(
r, bundles_activity=(*r.bundles_activity, name_id),
)
new_rails = (
instance.rails[:i] + (new_r,) + instance.rails[i + 1:]
)
return dataclasses.replace(instance, rails=new_rails)
raise KeyError(
f"Rail {reconciler_name!r} not found",
)
raise ValueError(
f"reconciler_kind={reconciler_kind!r} not recognized "
f"(expected 'transfer_template' or 'aggregating_rail')"
)
[docs]
def create_l2_entity(
instance: L2Instance,
kind: EntityKind,
fields: Mapping[str, Any], # typing-smell: ignore[explicit-any]: heterogeneous form-submitted field values; per-entity dataclass fields differ
) -> L2Instance:
"""Append a new entity to the kind's collection.
Builds the entity from ``fields`` (already coerced to dataclass-
field types by the caller). Required-but-missing fields raise
``ValueError`` from the dataclass constructor; ID collisions
raise ``ValueError`` here (we'd rather fail loud at construction
than let a duplicate slip into the collection and have the
validator's reference-resolution surface it as a confusing
indirect error).
Returns a new ``L2Instance``. The caller composes ``validate(...)``
afterward to catch L2-graph break (e.g., a Rail referencing roles
that don't exist yet).
"""
if kind == "account":
new_id = fields.get("id")
if not new_id:
raise ValueError("Account.id is required")
if any(str(a.id) == str(new_id) for a in instance.accounts):
raise ValueError(f"Account id {new_id!r} already exists")
new_acc = Account(
id=Identifier(str(new_id)),
scope=fields.get("scope") or "internal",
name=Name(str(fields["name"])) if fields.get("name") else None,
role=Identifier(str(fields["role"])) if fields.get("role") else None,
parent_role=(
Identifier(str(fields["parent_role"]))
if fields.get("parent_role") else None
),
expected_eod_balance=fields.get("expected_eod_balance"),
description=fields.get("description"),
)
return dataclasses.replace(
instance, accounts=(*instance.accounts, new_acc),
)
if kind == "account_template":
new_role = fields.get("role")
if not new_role:
raise ValueError("AccountTemplate.role is required")
if any(
str(t.role) == str(new_role) for t in instance.account_templates
):
raise ValueError(
f"AccountTemplate role {new_role!r} already exists",
)
new_t = AccountTemplate(
role=Identifier(str(new_role)),
scope=fields.get("scope") or "internal",
parent_role=(
Identifier(str(fields["parent_role"]))
if fields.get("parent_role") else None
),
expected_eod_balance=fields.get("expected_eod_balance"),
description=fields.get("description"),
instance_id_template=fields.get("instance_id_template"),
instance_name_template=fields.get("instance_name_template"),
)
return dataclasses.replace(
instance, account_templates=(*instance.account_templates, new_t),
)
if kind == "rail":
new_name = fields.get("name")
if not new_name:
raise ValueError("Rail.name is required")
if any(str(r.name) == str(new_name) for r in instance.rails):
raise ValueError(f"Rail name {new_name!r} already exists")
# X.4.f.11.5 — subtype dispatch. The studio editor's create
# page is 2-step: a picker page routes the operator to
# ``?subtype=two_leg|single_leg``, the form filters to that
# subtype's FieldSpecs, and the hidden ``subtype`` form input
# arrives here as a fields key. Construct the right
# discriminated-union arm. ``aggregating`` is shared; the
# form coerces "true"/"false" to bool.
subtype = fields.get("subtype")
aggregating = bool(fields.get("aggregating") or False)
new_r: Rail
# X.4.f.11.6/.7/.9 — coerce already-typed Tier-2 fields through.
# metadata_keys / posted_requirements / bundles_activity arrive
# as tuples (textarea-coerce / multi_select). aging Durations
# arrive as datetime.timedelta (loader's _load_duration).
metadata_keys_v = fields.get("metadata_keys") or ()
posted_requirements_v = fields.get("posted_requirements") or ()
bundles_activity_v = fields.get("bundles_activity") or ()
max_pending_age_v = fields.get("max_pending_age")
max_unbundled_age_v = fields.get("max_unbundled_age")
metadata_value_examples_v = fields.get("metadata_value_examples") or ()
if subtype == "single_leg":
leg_role_raw: object = fields.get("leg_role") or ()
leg_role: tuple[Identifier, ...] = (
tuple(
Identifier(str(x)) # pyright: ignore[reportUnknownArgumentType] # WHY: form-data tuple elements are untyped Any per Mapping[str, Any] contract; str() is the safe coerce boundary
for x in leg_role_raw # pyright: ignore[reportUnknownVariableType] # WHY: same untyped-Any per-element issue
)
if isinstance(leg_role_raw, (list, tuple))
else ()
)
leg_direction = fields.get("leg_direction")
if not leg_direction:
raise ValueError(
"SingleLegRail.leg_direction is required",
)
new_r = SingleLegRail(
name=Identifier(str(new_name)),
metadata_keys=metadata_keys_v, # pyright: ignore[reportArgumentType] # WHY: form-typed tuple[Identifier, ...] from coerce path
leg_role=leg_role,
leg_direction=leg_direction, # pyright: ignore[reportArgumentType] # WHY: form-coerced str; LegDirection is Literal["Debit","Credit","Variable"] — value validated by FieldSpec options + the enclosing validator
origin=(
str(fields["origin"]) if fields.get("origin") else None
),
posted_requirements=posted_requirements_v, # pyright: ignore[reportArgumentType] # WHY: form-typed tuple[Identifier, ...]
max_pending_age=max_pending_age_v, # pyright: ignore[reportArgumentType] # WHY: form-coerced via _load_duration → timedelta or None
max_unbundled_age=max_unbundled_age_v, # pyright: ignore[reportArgumentType] # WHY: form-coerced via _load_duration to timedelta or None
aggregating=aggregating,
bundles_activity=bundles_activity_v, # pyright: ignore[reportArgumentType] # WHY: form-typed tuple[Identifier, ...]
description=fields.get("description"),
metadata_value_examples=metadata_value_examples_v, # pyright: ignore[reportArgumentType] # WHY: form-typed nested tuple from yaml_block coerce
cadence=fields.get("cadence"), # pyright: ignore[reportArgumentType] # WHY: AI.2.a — form-coerced CadenceExpression or None (was dropped on create)
amount_typical_range=fields.get("amount_typical_range"), # pyright: ignore[reportArgumentType] # WHY: AI.2.a — form-coerced tuple[Money, Money] or None (was dropped on create)
firings_typical_per_period=fields.get("firings_typical_per_period"), # pyright: ignore[reportArgumentType] # WHY: AI.2.a — form-coerced FiringsTypicalPerPeriod or None (was dropped on create)
)
else:
# Default to two_leg when subtype is missing — caller is
# expected to set it for new rail creates, but keeping the
# default lets the legacy code path (no subtype provided)
# behave identically: an empty TwoLegRail validator rejects.
src_raw: object = fields.get("source_role") or ()
dst_raw: object = fields.get("destination_role") or ()
source_role: tuple[Identifier, ...] = (
tuple(
Identifier(str(x)) # pyright: ignore[reportUnknownArgumentType] # WHY: form-data tuple elements are untyped Any per Mapping[str, Any] contract
for x in src_raw # pyright: ignore[reportUnknownVariableType] # WHY: form-data tuple elements are untyped Any per contract
)
if isinstance(src_raw, (list, tuple))
else ()
)
destination_role: tuple[Identifier, ...] = (
tuple(
Identifier(str(x)) # pyright: ignore[reportUnknownArgumentType] # WHY: form-data tuple elements are untyped Any per Mapping[str, Any] contract
for x in dst_raw # pyright: ignore[reportUnknownVariableType] # WHY: form-data tuple elements are untyped Any per contract
)
if isinstance(dst_raw, (list, tuple))
else ()
)
new_r = TwoLegRail(
name=Identifier(str(new_name)),
metadata_keys=metadata_keys_v, # pyright: ignore[reportArgumentType] # WHY: form-typed tuple[Identifier, ...] from coerce path
source_role=source_role,
destination_role=destination_role,
origin=(
str(fields["origin"]) if fields.get("origin") else None
),
source_origin=(
str(fields["source_origin"])
if fields.get("source_origin") else None
),
destination_origin=(
str(fields["destination_origin"])
if fields.get("destination_origin") else None
),
expected_net=fields.get("expected_net"), # pyright: ignore[reportArgumentType] # WHY: form-coerced Decimal via money kind, or None
posted_requirements=posted_requirements_v, # pyright: ignore[reportArgumentType] # WHY: form-typed tuple[Identifier, ...]
max_pending_age=max_pending_age_v, # pyright: ignore[reportArgumentType] # WHY: form-coerced via _load_duration → timedelta or None
max_unbundled_age=max_unbundled_age_v, # pyright: ignore[reportArgumentType] # WHY: form-coerced via _load_duration to timedelta or None
aggregating=aggregating,
bundles_activity=bundles_activity_v, # pyright: ignore[reportArgumentType] # WHY: form-typed tuple[Identifier, ...]
description=fields.get("description"),
metadata_value_examples=metadata_value_examples_v, # pyright: ignore[reportArgumentType] # WHY: form-typed nested tuple from yaml_block coerce
cadence=fields.get("cadence"), # pyright: ignore[reportArgumentType] # WHY: AI.2.a — form-coerced CadenceExpression or None (was dropped on create)
amount_typical_range=fields.get("amount_typical_range"), # pyright: ignore[reportArgumentType] # WHY: AI.2.a — form-coerced tuple[Money, Money] or None (was dropped on create)
firings_typical_per_period=fields.get("firings_typical_per_period"), # pyright: ignore[reportArgumentType] # WHY: AI.2.a — form-coerced FiringsTypicalPerPeriod or None (was dropped on create)
)
return dataclasses.replace(instance, rails=(*instance.rails, new_r))
if kind == "transfer_template":
new_name = fields.get("name")
if not new_name:
raise ValueError("TransferTemplate.name is required")
if any(
str(t.name) == str(new_name)
for t in instance.transfer_templates
):
raise ValueError(
f"TransferTemplate name {new_name!r} already exists",
)
if not fields.get("completion"):
raise ValueError("TransferTemplate.completion is required")
if fields.get("expected_net") is None:
raise ValueError("TransferTemplate.expected_net is required")
# leg_rails comes from the multi_select form field as a tuple
# of Identifiers. Empty tuple is allowed at construction here
# but the caller's validate() rejects it (TransferTemplate must
# have at least one leg_rail) — surfaces inline on the create
# page as a 400 + the operator's selection preserved.
leg_rails_raw: object = fields.get("leg_rails") or ()
if isinstance(leg_rails_raw, (list, tuple)):
leg_rails: tuple[Identifier, ...] = tuple(
Identifier(str(r)) for r in leg_rails_raw # pyright: ignore[reportUnknownVariableType,reportUnknownArgumentType] # WHY: form-data values arrive as untyped strings; per-element coercion is the boundary
)
else:
leg_rails = (Identifier(str(leg_rails_raw)),)
# AB.3.7 — leg_rail_xor_groups arrives from
# ``multi_select_groups`` as ``tuple[tuple[Identifier, ...], ...]``.
# Defensive coerce: form-data path normalizes to Identifier
# before this point, but accept loose nested lists too for
# API-style POSTs / future call sites.
xor_raw: object = fields.get("leg_rail_xor_groups") or ()
if isinstance(xor_raw, (list, tuple)):
leg_rail_xor_groups: tuple[tuple[Identifier, ...], ...] = tuple(
tuple(
Identifier(str(r)) # pyright: ignore[reportUnknownArgumentType] # WHY: nested tuple element type isn't narrowed
for r in group # pyright: ignore[reportUnknownVariableType] # WHY: nested tuple element type isn't narrowed
)
for group in xor_raw # pyright: ignore[reportUnknownVariableType] # WHY: tuple element type isn't narrowed
if isinstance(group, (list, tuple))
)
else:
leg_rail_xor_groups = ()
# AI.2.b — transfer_key arrives from the textarea FieldSpec as a
# tuple[Identifier, ...] (coerced by _coerce_field), or None when
# the operator left it blank (validator R12 permits empty). Accept
# a loose list/scalar too for API-style POSTs, mirroring leg_rails.
# Was hardcoded to () — every corpus template declares a non-empty
# transfer_key, so the old hardcode meant no template round-tripped.
transfer_key_raw: object = fields.get("transfer_key") or ()
if isinstance(transfer_key_raw, (list, tuple)):
transfer_key: tuple[Identifier, ...] = tuple(
Identifier(str(k)) for k in transfer_key_raw # pyright: ignore[reportUnknownVariableType,reportUnknownArgumentType] # WHY: form-data values arrive as untyped strings; per-element coercion is the boundary
)
else:
transfer_key = (Identifier(str(transfer_key_raw)),)
new_tt = TransferTemplate(
name=Identifier(str(new_name)),
expected_net=Money(fields["expected_net"]),
completion=str(fields["completion"]),
leg_rails=leg_rails,
leg_rail_xor_groups=leg_rail_xor_groups,
transfer_key=transfer_key,
description=fields.get("description"),
# BB.3 — create-path field drop discovered via the dogfood
# round-trip (BB.3 driver create-new path passes the full
# reference TT including firings_typical_per_period; create
# was silently dropping it like AI.2.a fixed for rails).
firings_typical_per_period=fields.get(
"firings_typical_per_period",
), # pyright: ignore[reportArgumentType]: form-coerced FiringsTypicalPerPeriod or None
)
return dataclasses.replace(
instance,
transfer_templates=(*instance.transfer_templates, new_tt),
)
if kind == "chain":
# Z.A grammar collapse — a chain row is now (parent, children, description?).
# No required / xor_group flags. The studio editor's create form
# supplies a children-checkbox-group → fields["children"] is a list[str].
parent = fields.get("parent")
children_raw = fields.get("children")
if not parent:
raise ValueError("Chain.parent is required")
if not isinstance(children_raw, (list, tuple)):
raise ValueError(
"Chain.children must be a non-empty list of "
"rail / template names (singleton ⇒ required, "
"multi ⇒ XOR per Z.A grammar).",
)
# AI.2.a — _coerce_form's chain_children path already builds a
# tuple[ChainChildSpec, ...] carrying per-child fan_in +
# expected_parent_count (the same shape mutate_l2 consumes on edit),
# so consume it directly: create now supports per-child fan-in +
# mixed-cardinality (sasquatch's MerchantSettlementCycle shape).
# Earlier this branch re-read the specs as bare names and synthesized
# ONE chain-level fan_in onto every child, silently dropping per-child
# fan-in on create. Fall back to a name-list + optional chain-level
# fan_in for non-form (API) callers that pass plain strings.
children: tuple[ChainChildSpec, ...]
if all(isinstance(c, ChainChildSpec) for c in children_raw): # pyright: ignore[reportUnknownVariableType] # WHY: children_raw is Any-typed form data, narrowed to list/tuple above
children = tuple(children_raw) # pyright: ignore[reportUnknownArgumentType, reportUnknownVariableType] # WHY: each element confirmed ChainChildSpec by the all() guard
else:
fan_in_raw = fields.get("fan_in", False)
fan_in = bool(fan_in_raw) if fan_in_raw is not None else False
expected_raw = fields.get("expected_parent_count")
expected_parent_count: int | None = (
int(expected_raw)
if expected_raw is not None and expected_raw != "" else None
)
children = tuple(
ChainChildSpec(
name=Identifier(str(c)), # pyright: ignore[reportUnknownArgumentType] # WHY: API-path element is Any; str() narrows
fan_in=fan_in,
expected_parent_count=expected_parent_count,
)
for c in children_raw # pyright: ignore[reportUnknownVariableType] # WHY: Any-typed form data narrowed to list/tuple above
)
if not children:
raise ValueError(
"Chain.children must be non-empty (singleton ⇒ required, "
"multi ⇒ XOR per Z.A grammar).",
)
child_names = tuple(c.name for c in children)
# Check for duplicate row by (parent, sorted-children-tuple).
new_key = (str(parent), tuple(sorted(str(c) for c in child_names)))
for c in instance.chains:
existing_key = (
str(c.parent),
tuple(sorted(str(ch.name) for ch in c.children)),
)
if existing_key == new_key:
raise ValueError(
f"Chain row for parent={parent!r} with "
f"children={list(child_names)!r} already exists.",
)
new_ce = Chain(
parent=Identifier(str(parent)),
children=children,
description=fields.get("description"),
)
return dataclasses.replace(
instance, chains=(*instance.chains, new_ce),
)
if kind == "limit_schedule":
parent_role = fields.get("parent_role")
rail = fields.get("rail")
cap = fields.get("cap")
# AB.1 — direction defaults to Outbound (preserves legacy
# behavior + the loader's same default for unset YAML keys).
direction_raw = fields.get("direction") or "Outbound"
if direction_raw not in ("Outbound", "Inbound"):
raise ValueError(
f"LimitSchedule.direction={direction_raw!r}: must be "
f"'Outbound' or 'Inbound'",
)
if not parent_role or not rail or cap is None:
raise ValueError(
"LimitSchedule.parent_role / .rail / .cap "
"are required",
)
if any(
str(ls.parent_role) == str(parent_role)
and str(ls.rail) == str(rail)
and str(ls.direction) == str(direction_raw)
for ls in instance.limit_schedules
):
raise ValueError(
f"LimitSchedule {parent_role}::{rail}::{direction_raw} "
f"already exists",
)
new_ls = LimitSchedule(
parent_role=Identifier(str(parent_role)),
rail=RailName(str(rail)),
cap=Money(cap),
direction=direction_raw, # pyright: ignore[reportArgumentType]: validated against the LimitDirection Literal set just above
description=fields.get("description"),
)
return dataclasses.replace(
instance, limit_schedules=(*instance.limit_schedules, new_ls),
)
raise ValueError(f"Unknown entity kind: {kind!r}")
[docs]
def singleton_save_l2(
instance: L2Instance,
kind: EntityKind,
yaml_text: str,
) -> L2Instance:
"""X.4.f.12 — write a singleton attribute from a raw YAML block.
BXa.1 (2026-05-30) removed the ``persona`` singleton — its fields
promoted to top-level + nuked. Remaining singletons: ``theme``
(still YAML-textarea) + ``instance`` (new structured form per
BXa.1; the YAML-textarea path stays as a fallback).
Empty / blank YAML clears the attribute back to ``None``
(silent-fallback contract — N.4.k for theme; equivalent for
instance). Bad YAML / wrong shape raises ``ValueError`` (loader's
own exceptions inherit ``L2LoaderError`` which carries an
actionable message). The studio handler catches both and
re-renders the form with the operator's typed content + the
inline error.
"""
if kind not in SINGLETON_KINDS:
raise ValueError(
f"singleton_save_l2: kind {kind!r} is not a singleton kind",
)
import yaml # noqa: PLC0415 — lazy
from recon_gen.common.l2.loader import ( # noqa: PLC0415 — lazy to dodge cycle
L2LoaderError,
_load_description,
_load_optional_string,
_load_role_business_day_offsets,
_load_theme,
)
raw = yaml_text.strip()
if not raw:
# Empty ⇒ clear the singleton; silent-fallback takes over.
if kind == "theme":
return dataclasses.replace(instance, theme=None)
# AI.2.c + BXa.1 — instance settings: empty block clears the
# top-level fields (description + role_business_day_offsets +
# institution_name + institution_acronym).
return dataclasses.replace(
instance,
description=None,
role_business_day_offsets=None,
institution_name=None,
institution_acronym=None,
)
try:
parsed = yaml.safe_load(raw)
except yaml.YAMLError as exc:
raise ValueError(f"Invalid YAML: {exc}") from exc
if not isinstance(parsed, dict):
raise ValueError(
f"Expected a YAML map; got {type(parsed).__name__}",
)
parsed_map = cast("dict[str, object]", parsed)
try:
if kind == "theme":
new_theme = _load_theme(parsed_map, path=kind)
return dataclasses.replace(instance, theme=new_theme)
# AI.2.c + BXa.1 — instance settings: the block IS the full
# state of the editable top-level fields (an omitted key clears
# that field). Reuse the loader's per-field validators.
return dataclasses.replace(
instance,
description=_load_description(
parsed_map.get("description"), path="description",
),
role_business_day_offsets=_load_role_business_day_offsets(
parsed_map.get("role_business_day_offsets"),
path="role_business_day_offsets",
),
institution_name=_load_optional_string(
parsed_map.get("institution_name"),
path="institution_name",
),
institution_acronym=_load_optional_string(
parsed_map.get("institution_acronym"),
path="institution_acronym",
),
)
except L2LoaderError as exc:
raise ValueError(str(exc)) from exc
[docs]
def delete_l2_entity(
instance: L2Instance,
kind: EntityKind,
entity_id: str,
) -> L2Instance:
"""Remove one entity. Caller composes ``validate()`` to surface
structural breaks.
Per the SPEC: "Structural break = reject, don't auto-cascade."
Deleting a Rail that some TransferTemplate.leg_rails still
references leaves the model in a state the strict validator
rejects; the Studio PUT handler catches ``L2ValidationError`` and
returns 400 with the message inline.
Returns:
A new ``L2Instance`` with the matched entity removed.
Raises:
KeyError: no entity with that ``entity_id`` exists.
"""
_matched, idx, collection = _find_entity(instance, kind, entity_id)
new_collection = collection[:idx] + collection[idx + 1:]
return _replace_collection(instance, kind, new_collection)
# ---------------------------------------------------------------------------
# Entity lookup + collection swap
# ---------------------------------------------------------------------------
def _find_entity(
instance: L2Instance,
kind: EntityKind,
entity_id: str,
) -> "tuple[Any, int, tuple[Any, ...]]": # typing-smell: ignore[explicit-any]: per-kind union; the tuple element type narrows on the kind dispatch
"""Locate ``entity_id`` in the right collection. Returns
``(entity, index, collection)``. Raises KeyError on miss.
"""
if kind == "account":
for i, a in enumerate(instance.accounts):
if str(a.id) == entity_id:
return a, i, instance.accounts
elif kind == "account_template":
for i, t in enumerate(instance.account_templates):
if str(t.role) == entity_id:
return t, i, instance.account_templates
elif kind == "rail":
for i, r in enumerate(instance.rails):
if str(r.name) == entity_id:
return r, i, instance.rails
elif kind == "transfer_template":
for i, tt in enumerate(instance.transfer_templates):
if str(tt.name) == entity_id:
return tt, i, instance.transfer_templates
elif kind == "chain":
# Z.A grammar collapse — composite key now "<parent>::<sorted-children-csv>".
# Sorted so the address is stable across yaml round-trips even
# if the children list got re-ordered during an edit.
for i, ch in enumerate(instance.chains):
children_csv = ",".join(sorted(str(c.name) for c in ch.children))
if f"{ch.parent}::{children_csv}" == entity_id:
return ch, i, instance.chains
elif kind == "limit_schedule":
# Composite key: "<parent_role>::<rail>::<direction>" (AB.1).
# Backward-compat: a 2-part key (legacy "<parent_role>::<rail>")
# means direction="Outbound" — that's all pre-AB.1 schedules had.
parts = entity_id.split("::")
if len(parts) == 2:
parts = [*parts, "Outbound"]
if len(parts) == 3:
parent_role, rail, direction = parts
for i, ls in enumerate(instance.limit_schedules):
if (
str(ls.parent_role) == parent_role
and str(ls.rail) == rail
and str(ls.direction) == direction
):
return ls, i, instance.limit_schedules
raise KeyError(f"{kind} {entity_id!r} not found in instance")
def _replace_collection(
instance: L2Instance,
kind: EntityKind,
new_collection: "tuple[Any, ...]", # typing-smell: ignore[explicit-any]: per-kind union; dataclasses.replace narrows at the call site
) -> L2Instance:
"""Swap one collection on the L2Instance, return a new copy."""
field_name = {
"account": "accounts",
"account_template": "account_templates",
"rail": "rails",
"transfer_template": "transfer_templates",
"chain": "chains",
"limit_schedule": "limit_schedules",
}[kind]
return dataclasses.replace(instance, **{field_name: new_collection})
# ---------------------------------------------------------------------------
# Per-kind rename walkers
# ---------------------------------------------------------------------------
def _rename_role(
instance: L2Instance, old: Identifier, new: Identifier,
) -> L2Instance:
"""Rewrite every role-typed reference: Account.role / parent_role,
AccountTemplate.role / parent_role, Rail's source/destination/leg
roles, LimitSchedule.parent_role.
"""
accounts = tuple(_rename_account_roles(a, old, new) for a in instance.accounts)
account_templates = tuple(
_rename_account_template_roles(t, old, new)
for t in instance.account_templates
)
rails = tuple(_rename_rail_roles(r, old, new) for r in instance.rails)
limit_schedules = tuple(
_rename_limit_schedule_role(ls, old, new) for ls in instance.limit_schedules
)
return dataclasses.replace(
instance,
accounts=accounts,
account_templates=account_templates,
rails=rails,
limit_schedules=limit_schedules,
)
def _rename_account_roles(
a: Account, old: Identifier, new: Identifier,
) -> Account:
role = new if a.role == old else a.role
parent_role = new if a.parent_role == old else a.parent_role
if role is a.role and parent_role is a.parent_role:
return a
return dataclasses.replace(a, role=role, parent_role=parent_role)
def _rename_account_template_roles(
t: AccountTemplate, old: Identifier, new: Identifier,
) -> AccountTemplate:
role = new if t.role == old else t.role
parent_role = new if t.parent_role == old else t.parent_role
if role is t.role and parent_role is t.parent_role:
return t
return dataclasses.replace(t, role=role, parent_role=parent_role)
def _rename_rail_roles(
r: Rail, old: Identifier, new: Identifier,
) -> Rail:
if isinstance(r, TwoLegRail):
new_src = _rename_role_expression(r.source_role, old, new)
new_dst = _rename_role_expression(r.destination_role, old, new)
if new_src is r.source_role and new_dst is r.destination_role:
return r
return dataclasses.replace(
r, source_role=new_src, destination_role=new_dst,
)
# SingleLegRail
new_leg = _rename_role_expression(r.leg_role, old, new)
if new_leg is r.leg_role:
return r
return dataclasses.replace(r, leg_role=new_leg)
def _rename_role_expression(
re: tuple[Identifier, ...], old: Identifier, new: Identifier,
) -> tuple[Identifier, ...]:
rewritten = tuple(new if r == old else r for r in re)
return rewritten if rewritten != re else re
def _rename_limit_schedule_role(
ls: LimitSchedule, old: Identifier, new: Identifier,
) -> LimitSchedule:
if ls.parent_role == old:
return dataclasses.replace(ls, parent_role=new)
return ls
def _rename_rail(
instance: L2Instance, old: Identifier, new: Identifier,
) -> L2Instance:
"""Rewrite every Rail-name reference: TransferTemplate.leg_rails,
Rail.bundles_activity, Chain.parent / .children[i]. Also bumps the
Rail's own .name (the rename's anchor target).
"""
rails = tuple(
dataclasses.replace(r, name=new) if r.name == old
else _rename_rail_bundles(r, old, new)
for r in instance.rails
)
transfer_templates = tuple(
_rename_template_leg_rails(tt, old, new)
for tt in instance.transfer_templates
)
chains = tuple(_rename_chain_endpoint(c, old, new) for c in instance.chains)
return dataclasses.replace(
instance,
rails=rails,
transfer_templates=transfer_templates,
chains=chains,
)
def _rename_rail_bundles(
r: Rail, old: Identifier, new: Identifier,
) -> Rail:
rewritten = tuple(new if b == old else b for b in r.bundles_activity)
if rewritten == r.bundles_activity:
return r
return dataclasses.replace(r, bundles_activity=rewritten)
def _rename_template_leg_rails(
tt: TransferTemplate, old: Identifier, new: Identifier,
) -> TransferTemplate:
rewritten = tuple(new if r == old else r for r in tt.leg_rails)
if rewritten == tt.leg_rails:
return tt
return dataclasses.replace(tt, leg_rails=rewritten)
def _rename_chain_endpoint(
c: Chain, old: Identifier, new: Identifier,
) -> Chain:
"""Rewrite Chain.parent + each Chain.children[i].name when they
match old. AB.6 (per-child): children entries are now ChainChildSpec
— rename swaps the `.name` field while preserving `.fan_in` +
`.expected_parent_count`."""
parent = new if c.parent == old else c.parent
children = tuple(
dataclasses.replace(ch, name=new) if ch.name == old else ch
for ch in c.children
)
if parent is c.parent and children == c.children:
return c
return dataclasses.replace(c, parent=parent, children=children)
def _rename_transfer_template(
instance: L2Instance, old: Identifier, new: Identifier,
) -> L2Instance:
"""Rewrite every TransferTemplate-name reference: Rail.bundles_activity,
Chain.parent / .children[i]. Plus the template's own .name.
"""
transfer_templates = tuple(
dataclasses.replace(tt, name=new) if tt.name == old else tt
for tt in instance.transfer_templates
)
rails = tuple(
_rename_rail_bundles(r, old, new) for r in instance.rails
)
chains = tuple(_rename_chain_endpoint(c, old, new) for c in instance.chains)
return dataclasses.replace(
instance,
transfer_templates=transfer_templates,
rails=rails,
chains=chains,
)