feat(design-systems): the model, the parent chain, and the guard on it
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 7s
CI & Build / integration (push) Successful in 18s
CI & Build / TypeScript typecheck (push) Successful in 20s
CI & Build / Python tests (push) Successful in 42s
CI & Build / Build & push image (push) Successful in 27s

Milestone #254 step 1 (#2286). A design system becomes a record Scribe holds
rather than prose in a rulebook: a named set of tokens with an OPTIONAL parent,
so a family system carries the house style and an app system carries only what
it changes. Answering "what does this app alter?" is then `list its tokens` —
nothing to compute.

`parent_id` is the whole model. It replaces both an `always_on` flag (a family
system is one with no parent) and a subscription join table (a project points at
ONE system; the chain supplies the rest) — less schema than the rulebook shape
it mirrors.

Two decisions the task left open, settled here:

- **Token values are JSONB keyed by mode**, not `value_light`/`value_dark`
  columns. The deciding argument was not flexibility, it was ambiguity: in a
  child system an unset mode means "inherit", in a root it means "not
  mode-dependent", and as columns both are NULL and the resolver cannot tell
  them apart. As a map, resolution is `{**parent, **child}` at every level with
  no special case for roots. Against it: queryability — but nothing filters
  tokens by value in SQL, so that buys a query no caller makes.
- **`group_name` is free text, no CHECK enum.** Groupings are each design
  system's own vocabulary; a whitelist would bake one install's kit into the
  schema. No CHECK is introduced anywhere, so rule #36 does not fire.

The cascade lives in `services/design_cascade.py` as pure functions over a
`{id: parent_id}` map, importing nothing — which is what lets both the service
and `access.py` use it without a cycle, and lets a test state a whole hierarchy
in one literal. Cycles are refused on WRITE by walking up from the proposed
parent (the cheap direction), and survived on READ by a visited-set, because a
loop from a direct DB edit must truncate rather than hang.

ACL (rule #78) is deliberately asymmetric: owning a system grants write,
reaching one through a project you can see grants READ ONLY. An editor on a
shared project must not be able to rewrite the family system every other project
in that family resolves through.

Also renames `services/design_system.py` -> `design_rulebook_import.py`. It is
the #251 prose extractor, whose role is already scheduled to become a one-shot
importer (#2288), and leaving it one character away from the new
`design_systems.py` was a trap for every later session.

Rule #115 throughout: nothing seeds a system or implies a default. An install
with zero design systems is ordinary, not degraded.
This commit is contained in:
2026-07-30 16:54:41 -04:00
parent 4ca3ab02c4
commit 03b3998585
13 changed files with 1022 additions and 3 deletions
+271
View File
@@ -0,0 +1,271 @@
"""Design system + token persistence, and the guard on the parent chain.
Access goes through `services/access.py` (rule #78), where reaching a system via
a project you can see grants READ but never write — see
`get_design_system_permission` for why that asymmetry exists.
The cascade itself lives in `services/design_cascade.py` as pure functions. This
module is the part that needs a database: loading the shape of the hierarchy,
refusing writes that would break it, and storing tokens.
Nothing here seeds or implies a default system. An install with no design
systems is an ordinary install, and every caller must handle an empty list as
the normal case rather than a missing prerequisite.
"""
import logging
from datetime import datetime, timezone
from sqlalchemy import select
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
logger = logging.getLogger(__name__)
class DesignSystemCycle(ValueError):
"""A parent assignment that would close an inheritance loop.
Raised rather than returned as None because the two outcomes need different
answers: None already means "not found, or not yours", and a caller that
conflated them would show "no such design system" for what is really "that
parent is one of its own descendants". Routes map this to a 400.
"""
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.
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.
"""
rows = (
await session.execute(
select(DesignSystem.id, DesignSystem.parent_id).where(
DesignSystem.owner_user_id == user_id,
DesignSystem.deleted_at.is_(None),
)
)
).all()
return dict(rows)
# --- design systems ---------------------------------------------------------
async def create_design_system(
user_id: int,
title: str,
description: str | None = None,
parent_id: int | None = None,
) -> DesignSystem | None:
"""Create a system, with or without a parent.
Returns None when `parent_id` names a system the caller may not write —
which, per the ACL, means one they do not own.
"""
if parent_id is not None and not await access.can_write_design_system(
user_id, parent_id
):
return None
async with async_session() as session:
system = DesignSystem(
owner_user_id=user_id,
title=title.strip(),
description=description,
parent_id=parent_id,
)
session.add(system)
await session.commit()
await session.refresh(system)
return system
async def get_design_system(user_id: int, design_system_id: int) -> DesignSystem | 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
if not await access.can_read_design_system(user_id, design_system_id):
return None
return system
async def list_design_systems(user_id: int) -> list[DesignSystem]:
"""The caller's own systems, ordered by title. Empty is normal."""
async with async_session() as session:
rows = await session.execute(
select(DesignSystem)
.where(
DesignSystem.owner_user_id == user_id,
DesignSystem.deleted_at.is_(None),
)
.order_by(DesignSystem.title)
)
return list(rows.scalars().all())
async def update_design_system(
user_id: int, design_system_id: int, **fields: object
) -> DesignSystem | None:
"""Partial update. Raises DesignSystemCycle if `parent_id` would loop.
`parent_id` is handled apart from the other fields because None is a
meaningful value for it — "make this a root" — where for every other field
None means "leave alone". Callers signal it by passing the key at all.
"""
if not await access.can_write_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
if "parent_id" in fields:
parent_id = fields.pop("parent_id")
if parent_id is not None:
parent_id = int(parent_id)
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)
):
raise DesignSystemCycle(
f"Design system {design_system_id} cannot inherit from "
f"{parent_id}: that system already inherits from it."
)
system.parent_id = parent_id
for key, value in fields.items():
if key in ("title", "description") and value is not None:
setattr(system, key, value)
system.updated_at = datetime.now(timezone.utc)
await session.commit()
await session.refresh(system)
return system
async def delete_design_system(user_id: int, design_system_id: int) -> bool:
"""Soft-delete a system. Children survive as roots (the FK is SET NULL)."""
if not await access.can_write_design_system(user_id, design_system_id):
return False
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 False
system.deleted_at = datetime.now(timezone.utc)
await session.commit()
return True
# --- tokens -----------------------------------------------------------------
async def create_token(
user_id: int,
design_system_id: int,
name: str,
value_by_mode: dict | None = None,
group_name: str | None = None,
purpose: str | None = None,
order_index: int = 0,
) -> DesignToken | None:
if not await access.can_write_design_system(user_id, design_system_id):
return None
async with async_session() as session:
token = DesignToken(
design_system_id=design_system_id,
name=name.strip(),
# `or {}` and not the argument as given: the column is NOT NULL so
# that absence has exactly one spelling. Passing None here would
# otherwise store JSON null and reintroduce the second empty state.
value_by_mode=value_by_mode or {},
group_name=group_name,
purpose=purpose,
order_index=order_index,
)
session.add(token)
await session.commit()
await session.refresh(token)
return token
async def list_tokens(user_id: int, design_system_id: int) -> list[DesignToken]:
"""One system's OWN tokens — its override set, not its effective set.
Resolving the chain is step 2's job; this deliberately answers the narrower
question ("what does this system change?") that the model exists to make
free.
"""
if not await access.can_read_design_system(user_id, design_system_id):
return []
async with async_session() as session:
rows = await session.execute(
select(DesignToken)
.where(
DesignToken.design_system_id == design_system_id,
DesignToken.deleted_at.is_(None),
)
.order_by(DesignToken.order_index.asc(), DesignToken.name.asc())
)
return list(rows.scalars().all())
async def update_token(
user_id: int, token_id: int, **fields: object
) -> DesignToken | None:
allowed = {"name", "value_by_mode", "group_name", "purpose", "order_index"}
async with async_session() as session:
token = await session.get(DesignToken, token_id)
if token is None or token.deleted_at is not None:
return None
if not await access.can_write_design_system(user_id, token.design_system_id):
return None
for key, value in fields.items():
if key in allowed and value is not None:
setattr(token, key, value)
token.updated_at = datetime.now(timezone.utc)
await session.commit()
await session.refresh(token)
return token
async def delete_token(user_id: int, token_id: int) -> bool:
async with async_session() as session:
token = await session.get(DesignToken, token_id)
if token is None or token.deleted_at is not None:
return False
if not await access.can_write_design_system(user_id, token.design_system_id):
return False
token.deleted_at = datetime.now(timezone.utc)
await session.commit()
return True
# --- the project pointer ----------------------------------------------------
async def set_project_design_system(
user_id: int, project_id: int, design_system_id: int | None
) -> bool:
"""Point a project at a design system, or at nothing (None clears it).
Needs write on the project and READ on the system: pointing at a system is
consuming it, not changing it, so a system shared with you through another
project is a legitimate choice here.
"""
if not await access.can_write_project(user_id, project_id):
return False
if design_system_id is not None and not await access.can_read_design_system(
user_id, design_system_id
):
return False
async with async_session() as session:
project = await session.get(Project, project_id)
if project is None or project.deleted_at is not None:
return False
project.design_system_id = design_system_id
project.updated_at = datetime.now(timezone.utc)
await session.commit()
return True