FabledScribe/src/scribe/services/design_systems.py

"""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_stylesheet import (
    check_code_against_tokens,
    derivation_report,
    duplicate_values,
    render_stylesheet,
)
from scribe.services.design_starter_roles import (
    DEFAULT_TOKEN_PREFIX,
    starter_tokens,
)
from scribe.services.design_cascade import (
    ResolvedToken,
    ancestry,
    resolve_tokens,
    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, owner_user_id: int) -> dict[int, int | None]:
    """`{id: parent_id}` for one owner's live systems — the hierarchy's shape.

    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 == owner_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,
    guidance: str | None = None,
    parent_id: int | None = None,
    starter_role_groups: list[str] | None = None,
    token_prefix: str = DEFAULT_TOKEN_PREFIX,
) -> 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.

    `starter_role_groups` seeds the system with named, VALUELESS token roles
    (#2349) — the moment a role is missing is the moment a literal gets written
    instead, so the cheapest time to name them is now. Pass a list of group
    names to choose, `[]` for none, or None for none.

    None and `[]` deliberately mean the same thing here, unlike in
    `starter_tokens` where None means "all": creation must not seed 40 rows
    into a system whose caller never asked. Opting in is the caller's job, and
    the UI's default of everything-checked lives in the UI.
    """
    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,
            guidance=guidance,
            parent_id=parent_id,
        )
        session.add(system)
        await session.commit()
        await session.refresh(system)

        if starter_role_groups:
            for row in starter_tokens(starter_role_groups, prefix=token_prefix):
                session.add(DesignToken(design_system_id=system.id, **row))
            await session.commit()
        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, system.owner_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", "guidance") 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,
    rationale: str | None = None,
    supersedes: list | 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,
            rationale=rationale,
            # `or []` for the same reason as value_by_mode above: the column is
            # NOT NULL so absence has one spelling, and None would store JSON
            # null instead of an empty array.
            supersedes=supersedes or [],
            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 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 design_context(user_id: int, design_system_id: int) -> dict | None:
    """What a session needs to know about a design system, before it writes UI.

    This is the DELIVERY side of a design system, and it exists because storing
    one does not make a session aware of it. Rules get pushed into every session
    by the plugin's SessionStart hook; a design system had no such channel, so
    the standards were reachable only by an agent that already knew to go
    looking — which is the same silent failure as a token nobody declares.

    Guidance is chain-merged, ANCESTOR-FIRST, and that is the point rather than
    a convenience. A child system holds only what it CHANGES, so its own
    guidance describes a departure from a house style it never restates. Hand an
    agent the leaf alone and it builds against a fragment, with no signal that
    the rest exists.

    Tokens are summarised, not listed: the count and the group names are enough
    to know what the system covers, and the full set is one call away. Sending
    a hundred token values into every session start would crowd out the context
    it is meant to inform.

    None when the caller may not read the system.
    """
    tokens = await resolve_design_system(user_id, design_system_id)
    if tokens is None:
        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(DesignSystem).where(DesignSystem.id.in_(chain))
            )
        ).scalars().all()

    by_id = {s.id: s for s in rows}
    return {
        "id": system.id,
        "title": system.title,
        "description": system.description or "",
        # Outermost ancestor first, so the reader meets the house style before
        # the app's departures from it.
        "inherits_from": [
            by_id[sid].title for sid in reversed(chain[1:]) if sid in by_id
        ],
        "guidance": [
            {
                "design_system_id": sid,
                "title": by_id[sid].title,
                "guidance": (by_id[sid].guidance or "").strip(),
            }
            for sid in reversed(chain)
            if sid in by_id and (by_id[sid].guidance or "").strip()
        ],
        "token_count": len(tokens),
        "token_groups": sorted({t.group_name for t in tokens if t.group_name}),
    }


async def update_token(
    user_id: int, token_id: int, **fields: object
) -> DesignToken | None:
    allowed = {
        "name", "value_by_mode", "group_name", "purpose", "rationale",
        "supersedes", "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


# --- the master sheet -------------------------------------------------------

async def stylesheet_for_system(
    user_id: int, design_system_id: int, root_selector: str = ":root"
) -> dict | None:
    """The master CSS sheet a design system generates, plus its reuse report.

    Purpose tokens only — the sheet styles no elements. See
    `services/design_stylesheet.py` for why that split is the design rather than
    a shortcut.

    Returns None if the caller may not read the system. The `duplicates` half is
    advisory: two tokens sharing a value are either a deliberate alias or one
    idea recorded twice, and only the operator knows which.
    """
    resolved = await resolve_design_system(user_id, design_system_id)
    if resolved is None:
        return None
    system = await get_design_system(user_id, design_system_id)

    return {
        "design_system_id": design_system_id,
        "css": render_stylesheet(
            resolved,
            root_selector=root_selector,
            title=system.title if system else "",
            design_system_id=design_system_id,
        ),
        "token_count": len(resolved),
        "valueless": [t.name for t in resolved if not t.value_by_mode],
        "duplicates": duplicate_values(resolved),
        # Formulas: which tokens are computed from others, and which of those
        # point at nothing. A broken formula is dropped by the browser without
        # any error, so the sheet cannot show it for itself.
        "derivation": derivation_report(resolved),
    }


async def check_snippets_against_system(
    user_id: int, design_system_id: int, project_id: int = 0
) -> dict | None:
    """Which recorded snippets disagree with this design system's sheet.

    The relation the operator named — "the snippets use the tags from the sheet"
    — turned into a check. For each snippet: `var(--x)` references with no such
    token, literals the sheet says to stop writing, and custom properties the
    snippet mints for itself instead of using shared ones.

    Snippets with nothing to report are omitted entirely. A list of everything
    that is fine is a list nobody reads twice.
    """
    resolved = await resolve_design_system(user_id, design_system_id)
    if resolved is None:
        return None

    from scribe.services import snippets as snippets_svc

    # `list_snippets` returns (rows, total) and caps limit at 100; `project_id`
    # must be None — not 0 — to reach across every project, since 0 would filter
    # to a project with that id.
    rows, _total = await snippets_svc.list_snippets(
        user_id=user_id, project_id=project_id or None, limit=100,
    )

    findings: list[dict] = []
    for row in rows:
        snippet_id = row.get("id")
        if snippet_id is None:
            continue
        # The list rows carry a preview, not the code. The check has to read the
        # whole body or it would report on a truncation.
        note = await snippets_svc.get_snippet(user_id=user_id, snippet_id=int(snippet_id))
        if note is None:
            continue
        report = check_code_against_tokens(note.body or "", resolved)
        if not (
            report["unknown"] or report["superseded_literals"] or report["local_definitions"]
        ):
            continue
        findings.append({
            "snippet_id": int(snippet_id),
            "title": note.title or "",
            **report,
        })

    return {
        "design_system_id": design_system_id,
        "checked": len(rows),
        "findings": findings,
    }