feat(design-systems): resolve the chain, and keep the argument not the verdict
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Successful in 15s
CI & Build / TypeScript typecheck (push) Successful in 33s
CI & Build / Python tests (push) Successful in 42s
CI & Build / Build & push image (push) Successful in 29s

Milestone #254 step 2 (#2287). `resolve_tokens` flattens a system's inheritance
chain into its effective token set — walk to the root, deepest wins by token
name. Pure and duck-typed, so a test states a whole hierarchy in literals and
the service hands the same function ORM rows.

**Provenance is stored as the contest, not the winner.** A ResolvedToken carries
every system that offered a value, per mode, deepest first — `[0]` won and
`[1:]` are what it shadowed. "Which system supplied this?" and "what did it
override?" are then two reads of one list and cannot disagree, where a winner
plus a separate provenance field would be two things to keep in step.

**Merging is per (name, MODE), and that is the storage decision paying off.** A
system that deepens one accent for light backgrounds while leaving dark alone
owns `base` and still inherits `dark`. A token-level "overridden here" flag
would have to lie about one of them, and the two-column shape could not have
represented it at all.

Metadata cascades separately by the same deepest-wins rule, with one exception:
`order_index` treats 0 as UNSTATED rather than "first", because 0 is the column
default. Reading it as a real value would let a colour-only override drag its
token to the top of its group — a visible reshuffle in return for a change that
touched nothing structural.

One fix to step 1 while wiring this up: `_parent_map` is now scoped to the
SYSTEM'S OWNER rather than the caller. A caller reading through a shared project
owns no link in the chain, so the caller-scoped version would have handed them
an empty forest and truncated the cascade to a single system — a page rendering
with plausible wrong values and no error anywhere. The ACL already grants read
along the whole chain; this is the loading side keeping that promise, and it now
has a test naming the shared-project case.

`BASE_MODE` moves from the model to the cascade module, where it belongs: it is
a resolution rule, not a storage fact, and design_cascade.py deliberately
imports nothing so both access.py and the service can depend on it.
This commit is contained in:
2026-07-30 17:02:31 -04:00
parent 03b3998585
commit 839d6902ad
6 changed files with 476 additions and 18 deletions
+150 -1
View File
@@ -9,7 +9,13 @@ Being pure is also what makes the cascade rule testable without a database:
these take a plain `{id: parent_id}` map, so a test states the shape of a
hierarchy in one literal instead of building one.
"""
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
# The mode key that applies when no more specific one does. Emitted CSS puts it
# on the base selector and every other key on a mode selector, mirroring how a
# stylesheet is actually written: light on `:root`, dark layered over it.
BASE_MODE = "base"
def ancestry(system_id: int, parents: Mapping[int, int | None]) -> list[int]:
@@ -59,3 +65,146 @@ def would_cycle(
if proposed_parent_id == system_id:
return True
return system_id in ancestry(proposed_parent_id, parents)
# ---------------------------------------------------------------------------
# Resolution — flattening a chain into an effective token set
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Contribution:
"""One system's offer for one token in one mode."""
system_id: int
value: str
@dataclass(frozen=True)
class ResolvedToken:
"""A token after the cascade, carrying the whole argument rather than the verdict.
`contributions` holds every system that supplied a value, per mode, DEEPEST
FIRST — so `[0]` is the winner and `[1:]` are what it shadowed. Storing the
contest rather than a winner plus a separate provenance field means there is
nothing to keep in sync: "which system supplied this?" and "what did it
override?" are both reads of the same list, and they cannot disagree.
Provenance is per MODE, not per token, because overriding is. A system that
deepens one accent for light backgrounds while leaving dark alone owns the
base value and inherits the dark one, and a token-level "overridden here"
flag would have to lie about one of them.
"""
name: str
contributions: dict[str, tuple[Contribution, ...]]
group_name: str | None
purpose: str | None
order_index: int
@property
def value_by_mode(self) -> dict[str, str]:
"""The effective value for each mode — the winner of each contest."""
return {mode: entries[0].value for mode, entries in self.contributions.items()}
@property
def origin_by_mode(self) -> dict[str, int]:
"""Which system supplied each mode's effective value."""
return {mode: entries[0].system_id for mode, entries in self.contributions.items()}
def value_for(self, mode: str) -> str | None:
"""The value to render in `mode`, falling back to the base mode.
This is the read rule the storage shape implies: a token that is not
mode-dependent carries only `base`, and asking it for "dark" must yield
the base value rather than nothing.
"""
entries = self.contributions.get(mode) or self.contributions.get(BASE_MODE)
return entries[0].value if entries else None
def is_overridden_in(self, system_id: int) -> bool:
"""Does `system_id` win any mode of this token AND shadow something?
The distinction the UI needs: a token this system introduced is not an
override, and a token it merely inherits is not either.
"""
return any(
len(entries) > 1 and entries[0].system_id == system_id
for entries in self.contributions.values()
)
def _sort_key(token: ResolvedToken) -> tuple:
# Ungrouped tokens sort last rather than first: a design system that has
# started grouping should read as its groups, with the not-yet-filed
# remainder at the end.
return (token.group_name is None, token.group_name or "", token.order_index, token.name)
def resolve_tokens(
system_id: int,
parents: Mapping[int, int | None],
tokens_by_system: Mapping[int, Sequence],
) -> list[ResolvedToken]:
"""Flatten a system's inheritance chain into its effective token set.
Walks from `system_id` up to the root and applies tokens by name, deepest
winning. That is the CSS cascade — precedence by name along a parent chain —
rather than an analogy to it, which is why the storage model and the
stylesheet model came out the same shape.
`tokens_by_system` maps a system id to its own token rows. Any object with
`.name`, `.value_by_mode`, `.group_name`, `.purpose` and `.order_index` will
do, so a test can state a hierarchy in literals and the service can pass ORM
rows to the same function.
The result includes tokens the system never mentions — inheriting one is
what puts it in the effective set. A system with no tokens of its own
resolves to its parent's set entire, which is the correct answer for an app
that has not departed from the family yet.
Merging is per (name, MODE): a child that supplies only a dark value
overrides only dark and keeps inheriting base. Metadata (`group_name`,
`purpose`, `order_index`) cascades separately by the same deepest-wins rule,
since a child overriding a value routinely leaves the family's description
of what the token is FOR untouched — and inheriting it beats blanking it.
"""
chain = ancestry(system_id, parents)
contributions: dict[str, dict[str, list[Contribution]]] = {}
metadata: dict[str, dict[str, object]] = {}
# Deepest first, so the first contribution seen for a (name, mode) wins and
# every later one is a shadowed ancestor appended behind it.
for depth_system_id in chain:
for token in tokens_by_system.get(depth_system_id) or ():
per_mode = contributions.setdefault(token.name, {})
for mode, value in (token.value_by_mode or {}).items():
per_mode.setdefault(mode, []).append(
Contribution(system_id=depth_system_id, value=value)
)
meta = metadata.setdefault(
token.name, {"group_name": None, "purpose": None, "order_index": None}
)
for field in ("group_name", "purpose"):
if meta[field] is None:
meta[field] = getattr(token, field, None)
# order_index alone treats 0 as UNSTATED rather than "first",
# because 0 is the column default. Reading it as a real value would
# let any child override drag its token to the top of the group and
# lose the family's ordering — a visible reshuffle in return for a
# change that only touched a colour.
if not meta["order_index"]:
meta["order_index"] = getattr(token, "order_index", 0) or None
resolved = [
ResolvedToken(
name=name,
contributions={
mode: tuple(entries) for mode, entries in per_mode.items()
},
group_name=metadata[name]["group_name"],
purpose=metadata[name]["purpose"],
order_index=metadata[name]["order_index"] or 0,
)
for name, per_mode in contributions.items()
]
return sorted(resolved, key=_sort_key)
+62 -8
View File
@@ -21,7 +21,12 @@ from scribe.models import async_session
from scribe.models.design_system import DesignSystem, DesignToken
from scribe.models.project import Project
from scribe.services import access
from scribe.services.design_cascade import would_cycle
from scribe.services.design_cascade import (
ResolvedToken,
ancestry,
resolve_tokens,
would_cycle,
)
logger = logging.getLogger(__name__)
@@ -36,17 +41,23 @@ class DesignSystemCycle(ValueError):
"""
async def _parent_map(session, user_id: int) -> dict[int, int | None]:
"""`{id: parent_id}` for the caller's live systems — the hierarchy's shape.
async def _parent_map(session, owner_user_id: int) -> dict[int, int | None]:
"""`{id: parent_id}` for one owner's live systems — the hierarchy's shape.
Owner-scoped, which is also the constraint on parenting: you can only build
a chain out of systems you own. A borrowed link would let someone else's
delete or re-parent silently restyle your app.
Scoped to the OWNER of the systems, not the caller, and the distinction is
load-bearing on the read path: a caller reading through a shared project
does not own any link in the chain, and a caller-scoped map would hand them
an empty forest and a cascade truncated to one system. They would get a page
that renders with plausible wrong values and no error anywhere.
Owner-scoping is also the constraint on parenting: a chain may only be built
from systems its owner controls, or someone else's delete or re-parent would
silently restyle your app.
"""
rows = (
await session.execute(
select(DesignSystem.id, DesignSystem.parent_id).where(
DesignSystem.owner_user_id == user_id,
DesignSystem.owner_user_id == owner_user_id,
DesignSystem.deleted_at.is_(None),
)
)
@@ -131,7 +142,9 @@ async def update_design_system(
if not await access.can_write_design_system(user_id, parent_id):
return None
if would_cycle(
design_system_id, parent_id, await _parent_map(session, user_id)
design_system_id,
parent_id,
await _parent_map(session, system.owner_user_id),
):
raise DesignSystemCycle(
f"Design system {design_system_id} cannot inherit from "
@@ -213,6 +226,47 @@ async def list_tokens(user_id: int, design_system_id: int) -> list[DesignToken]:
return list(rows.scalars().all())
async def resolve_design_system(
user_id: int, design_system_id: int
) -> list[ResolvedToken] | None:
"""A system's EFFECTIVE token set — everything it inherits, with its own on top.
None when the caller may not read the system; an empty list when the chain
genuinely holds no tokens, which is an ordinary state for a system that has
just been created.
Two queries regardless of how deep the chain runs: one for the hierarchy's
shape, one for every token in it. The flattening itself is
`design_cascade.resolve_tokens` — pure, so the cascade rule is tested
without a database and this function is only the loading.
"""
if not await access.can_read_design_system(user_id, design_system_id):
return None
async with async_session() as session:
system = await session.get(DesignSystem, design_system_id)
if system is None or system.deleted_at is not None:
return None
parents = await _parent_map(session, system.owner_user_id)
chain = ancestry(design_system_id, parents)
rows = (
await session.execute(
select(DesignToken)
.where(
DesignToken.design_system_id.in_(chain),
DesignToken.deleted_at.is_(None),
)
.order_by(DesignToken.order_index.asc(), DesignToken.name.asc())
)
).scalars().all()
tokens_by_system: dict[int, list[DesignToken]] = {}
for token in rows:
tokens_by_system.setdefault(token.design_system_id, []).append(token)
return resolve_tokens(design_system_id, parents, tokens_by_system)
async def update_token(
user_id: int, token_id: int, **fields: object
) -> DesignToken | None: