FabledScribe/src/scribe/services/rulebooks.py

"""Rulebook / topic / rule service layer — single source of truth used by
both routes/rulebooks.py and mcp/tools/rulebooks.py.

Ownership enforcement: every function takes user_id and scopes through
rulebooks.owner_user_id. Functions return models or model.to_dict() output
depending on the caller's needs (mirroring services/events.py pattern).
"""
from __future__ import annotations

import logging
from collections.abc import Iterable
from datetime import datetime
from typing import Optional

from sqlalchemy import and_, delete as sql_delete, insert, or_, select

from scribe.models import async_session
from scribe.models.system import System
from scribe.models.rulebook import Rulebook
from scribe.services.verification import (
    days_since_verified as _days_since_verified,
    last_verified_label as _last_verified_label,
)
from scribe.services import rule_versions
from scribe.models.rule_version import RuleVersion
from scribe.services.rule_usage import record_rule_surfaced

logger = logging.getLogger(__name__)


# ── Rulebook CRUD ────────────────────────────────────────────────────────

async def create_rulebook(
    user_id: int, title: str, description: str = "",
) -> Rulebook:
    """Create a new rulebook owned by user_id. Returns the persisted model."""
    async with async_session() as session:
        rb = Rulebook(
            owner_user_id=user_id, title=title, description=description or None,
        )
        session.add(rb)
        await session.commit()
        await session.refresh(rb)
        return rb


async def list_rulebooks(user_id: int) -> list[Rulebook]:
    """List rulebooks owned by user_id, ordered by title."""
    async with async_session() as session:
        result = await session.execute(
            select(Rulebook)
            .where(Rulebook.owner_user_id == user_id, Rulebook.deleted_at.is_(None))
            .order_by(Rulebook.title)
        )
        return list(result.scalars().all())


async def get_rulebook(rulebook_id: int, user_id: int) -> Optional[Rulebook]:
    """Get a rulebook by id, scoped to user_id. None if not owned or not found."""
    async with async_session() as session:
        result = await session.execute(
            select(Rulebook).where(
                Rulebook.id == rulebook_id,
                Rulebook.owner_user_id == user_id,
                Rulebook.deleted_at.is_(None),
            )
        )
        return result.scalar_one_or_none()


async def update_rulebook(
    rulebook_id: int, user_id: int, **fields,
) -> Optional[Rulebook]:
    """Partial update. Returns updated rulebook or None if not found."""
    async with async_session() as session:
        result = await session.execute(
            select(Rulebook).where(
                Rulebook.id == rulebook_id,
                Rulebook.owner_user_id == user_id,
            )
        )
        rb = result.scalar_one_or_none()
        if rb is None:
            return None
        allowed = {"title", "description", "always_on"}
        for key, value in fields.items():
            if key in allowed and value is not None:
                setattr(rb, key, value)
        await session.commit()
        await session.refresh(rb)
        return rb


async def delete_rulebook(rulebook_id: int, user_id: int) -> None:
    """Delete a rulebook. Cascade-deletes topics, rules, subscriptions."""
    async with async_session() as session:
        result = await session.execute(
            select(Rulebook).where(
                Rulebook.id == rulebook_id,
                Rulebook.owner_user_id == user_id,
            )
        )
        rb = result.scalar_one_or_none()
        if rb is None:
            return
        await session.delete(rb)
        await session.commit()


async def find_rulebook_by_title(
    user_id: int, title: str,
) -> Optional[Rulebook]:
    """Used by the port script for the dupe-guard. None if not found."""
    async with async_session() as session:
        result = await session.execute(
            select(Rulebook).where(
                Rulebook.owner_user_id == user_id,
                Rulebook.title == title,
                Rulebook.deleted_at.is_(None),
            )
        )
        return result.scalar_one_or_none()


# ── Topic CRUD ──────────────────────────────────────────────────────────

from scribe.models.rulebook import RulebookTopic


async def _assert_rulebook_owned(session, rulebook_id: int, user_id: int) -> None:
    """Raise ValueError if rulebook doesn't exist or isn't owned by user.
    Centralizes ownership check used by all topic/rule operations.
    """
    result = await session.execute(
        select(Rulebook).where(
            Rulebook.id == rulebook_id,
            Rulebook.owner_user_id == user_id,
            Rulebook.deleted_at.is_(None),
        )
    )
    if result.scalar_one_or_none() is None:
        raise ValueError(f"rulebook {rulebook_id} not found")


async def create_topic(
    rulebook_id: int, user_id: int, title: str,
    description: str = "", order_index: int = 0,
) -> RulebookTopic:
    async with async_session() as session:
        await _assert_rulebook_owned(session, rulebook_id, user_id)
        topic = RulebookTopic(
            rulebook_id=rulebook_id,
            title=title,
            description=description or None,
            order_index=order_index,
        )
        session.add(topic)
        await session.commit()
        await session.refresh(topic)
        return topic


async def list_topics(rulebook_id: int, user_id: int) -> list[RulebookTopic]:
    async with async_session() as session:
        await _assert_rulebook_owned(session, rulebook_id, user_id)
        result = await session.execute(
            select(RulebookTopic)
            .where(
                RulebookTopic.rulebook_id == rulebook_id,
                RulebookTopic.deleted_at.is_(None),
            )
            .order_by(RulebookTopic.order_index, RulebookTopic.title)
        )
        return list(result.scalars().all())


async def get_topic(topic_id: int, user_id: int) -> Optional[RulebookTopic]:
    """Get a topic, scoped via the rulebook owner."""
    async with async_session() as session:
        result = await session.execute(
            select(RulebookTopic)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .where(
                RulebookTopic.id == topic_id,
                Rulebook.owner_user_id == user_id,
                RulebookTopic.deleted_at.is_(None),
                Rulebook.deleted_at.is_(None),
            )
        )
        return result.scalar_one_or_none()


async def update_topic(
    topic_id: int, user_id: int, **fields,
) -> Optional[RulebookTopic]:
    async with async_session() as session:
        result = await session.execute(
            select(RulebookTopic)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .where(
                RulebookTopic.id == topic_id,
                Rulebook.owner_user_id == user_id,
            )
        )
        topic = result.scalar_one_or_none()
        if topic is None:
            return None
        allowed = {"title", "description", "order_index"}
        for key, value in fields.items():
            if key in allowed and value is not None:
                setattr(topic, key, value)
        await session.commit()
        await session.refresh(topic)
        return topic


async def delete_topic(topic_id: int, user_id: int) -> None:
    async with async_session() as session:
        result = await session.execute(
            select(RulebookTopic)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .where(
                RulebookTopic.id == topic_id,
                Rulebook.owner_user_id == user_id,
            )
        )
        topic = result.scalar_one_or_none()
        if topic is None:
            return
        await session.delete(topic)
        await session.commit()


# ── Rule CRUD ──────────────────────────────────────────────────────────

from scribe.models.rulebook import Rule, RuleRelation, rule_systems


async def _assert_topic_owned(session, topic_id: int, user_id: int) -> None:
    """Raise ValueError if topic doesn't exist or isn't in user's rulebook."""
    result = await session.execute(
        select(RulebookTopic)
        .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
        .where(
            RulebookTopic.id == topic_id,
            Rulebook.owner_user_id == user_id,
            RulebookTopic.deleted_at.is_(None),
            Rulebook.deleted_at.is_(None),
        )
    )
    if result.scalar_one_or_none() is None:
        raise ValueError(f"topic {topic_id} not found")


async def _assert_project_owned(session, project_id: int, user_id: int) -> None:
    """Raise ValueError if project doesn't exist or isn't owned by user."""
    from scribe.models.project import Project
    result = await session.execute(
        select(Project).where(
            Project.id == project_id,
            Project.user_id == user_id,
            Project.deleted_at.is_(None),
        )
    )
    if result.scalar_one_or_none() is None:
        raise ValueError(f"project {project_id} not found")


async def _assert_rulebook_rule_owned(session, rule_id: int, user_id: int) -> None:
    """Raise ValueError if rule isn't a rulebook rule the user owns.

    Project-scoped rules (Rule.project_id set, topic_id NULL) are NOT
    suppressible — they belong to the project; delete them instead. This
    helper deliberately excludes them.
    """
    from scribe.models.rulebook import Rule
    result = await session.execute(
        select(Rule)
        .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
        .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
        .where(
            Rule.id == rule_id,
            Rulebook.owner_user_id == user_id,
            Rule.deleted_at.is_(None),
            RulebookTopic.deleted_at.is_(None),
            Rulebook.deleted_at.is_(None),
        )
    )
    if result.scalar_one_or_none() is None:
        raise ValueError(f"rule {rule_id} not found or not a rulebook rule")


# The vocabularies migration 0088's CHECK constraints enforce. Named here so
# a caller can be corrected before the database refuses it (rule 36 keeps the
# two in step; this keeps the error readable).
TIERS = ("always_on", "conditional")
RELATION_KINDS = ("co_surfaces", "overrides", "elaborates")
# Migration 0098's CHECK. `rule` binds; `preference` is how the operator
# wants work done — see the model comment for why both live on one table.
KINDS = ("rule", "preference")


# The rule columns that are nullable, and therefore the ones where EMPTY has
# to mean empty. A write that stores "" leaves a column that is not NULL and
# not content — `verify_with IS NOT NULL` would then be true for a rule with
# no check, and the staleness sweep would list rules it should never see.
# Normalising here, at the one service seam, is what makes "unset" a single
# state instead of two that read alike through to_dict's `or ""`.
NULLABLE_RULE_TEXT = (
    "why", "how_to_apply", "when_to_apply", "verify_with", "expires_when",
)


def _valid_tier(tier: str) -> str:
    """An unrecognised tier falls back to always_on — the SAFE direction.

    Getting this wrong the other way would silently stop a rule binding, which
    is the one failure this whole milestone exists to prevent. A rule that
    preloads when it did not need to costs context; a rule that quietly stops
    preloading costs the behaviour it was written for.
    """
    return tier if tier in TIERS else "always_on"


def _valid_kind(kind: str) -> str:
    """An unrecognised kind falls back to `rule` — the SAFE direction.

    Same shape as _valid_tier and the same argument, pointed at force instead
    of delivery. A preference wrongly treated as binding costs a little
    friction: the reader is told something is required that was only
    preferred. A rule wrongly treated as a preference costs the thing the rule
    was written to prevent, and costs it silently, because nothing downstream
    can tell a softened rule from a preference that was always one.

    Between a reader who is too careful and a reader who is not careful
    enough, the typo should produce the first.
    """
    return kind if kind in KINDS else "rule"


# Re-exported, not redefined. Notes gained the same trio in milestone 317 and
# this reading of it is genuinely common, so it moved to services/verification
# — the DRY win note 3163 names, as against sharing the QUERY, which the two
# record types cannot (a rule scopes by rulebook ownership, a note by the note
# ACL). Kept importable from here because callers already reach for it here.
last_verified_label = _last_verified_label


def rule_brief(rule: Rule, **extra) -> dict:
    """The shape a rule takes when it is SURFACED rather than opened.

    One builder for every payload that hands rules to an agent, because there
    were three copies of this dict and they had already diverged — two carried
    `topic_id`, one didn't, and none carried the timestamps the model has held
    all along. That omission is why a rule written before the capability it
    duplicates was indistinguishable, at read time, from one still doing work
    (the FabledCurator case, note 3026).

    `updated_at` is a DATE, not a stamp: the question it answers is "how old
    is this?", and a full ISO string across an always-on set is ~2k characters
    of payload for a precision nobody reads.

    `why` and `how_to_apply` are deliberately NOT here — they are the depth a
    caller gets from get_rule, and putting them in every listing is the bloat
    this milestone is about.
    """
    out = {
        "id": rule.id,
        "title": rule.title,
        "statement": rule.statement,
        "topic_id": rule.topic_id,
        "tier": rule.tier,
        # Unconditional, and the payload cost is accepted deliberately. Every
        # other optional key below is attached only when present, because an
        # absent key should never read as a capability the record lacks. Force
        # is the opposite case: a reader seeing no `kind` would have to assume
        # one, and the assumption it would reach for — "this binds" — is the
        # expensive one to get wrong in the other direction. Say it outright.
        "kind": rule.kind or "rule",
        "updated_at": rule.updated_at.date().isoformat() if rule.updated_at else None,
    }
    # Attached only when present (#2483: never a null key that reads as a
    # capability the record doesn't have).
    if rule.when_to_apply:
        out["when_to_apply"] = rule.when_to_apply
    if rule.arose_from_id:
        out["arose_from_id"] = rule.arose_from_id
    # Present ONLY on a rule that carries a check — its presence is the
    # signal, and it says two things at once: this rule asserts a fact that
    # can go false, and here is how long ago anyone confirmed it. The check
    # text itself stays in get_rule; a listing needs to know WHICH rules can
    # rot, not how to test them. "never" rather than null, per #2483: a key
    # that reads as an unused capability is a different claim from a rule
    # nobody has ever verified.
    stamp = last_verified_label(rule)
    if stamp:
        out["last_verified"] = stamp
    out.update({k: v for k, v in extra.items() if v is not None})
    return out


def _refresh_rule_embedding(rule: Rule) -> None:
    """Re-index a rule after a write. Fire-and-forget, like the note twin.

    Lazy import so this module doesn't pull in the embedder; every exception
    swallowed because a rule that SAVED must not fail on its index refresh —
    a stale vector costs a missed search hit, a raised exception costs the
    write. No running loop (unit tests, scripts) is ordinary, not an error.

    Detaching also means this task races anything that deletes the rule out
    from under it. That is not handled here: `upsert_rule_embedding` claims
    the rule's row before touching its vectors, and loses if it can't (#3262).
    """
    try:
        import asyncio

        from scribe.services.embeddings import upsert_rule_embedding

        asyncio.create_task(
            upsert_rule_embedding(
                rule.id, rule.title, rule.statement, rule.when_to_apply,
            )
        )
    except RuntimeError:
        pass          # no running loop — a sync caller, not a failure
    except Exception:  # noqa: BLE001 - never let indexing break a write
        logger.exception("embedding refresh failed for rule %s", rule.id)


async def co_surfaced_partners(
    user_id: int, rule_ids: list[int], exclude_ids: set[int] | None = None,
) -> list[Rule]:
    """Rules that must arrive WITH the given ones, because they fail together.

    This is the whole reason `co_surfaces` exists. Rule 144 was split off rule
    46 and folded back into it the same day, on the correct observation that
    "either rule could surface without the other and miss exposing a project to
    what the entire shape is intended to be." Merging was the only fix
    available; this is the fix that should have been available.

    Two limits, both deliberate:

    - Only rules the caller OWNS. An edge is not a back door into someone
      else's rulebook.
    - `exclude_ids` is honoured, and callers pass the project's SUPPRESSIONS.
      A project that explicitly muted a rule should not have it dragged back in
      by an edge — the suppression is a decision, and the edge does not
      outrank it.
    """
    if not rule_ids:
        return []
    known = set(rule_ids) | (exclude_ids or set())
    async with async_session() as session:
        edges = (await session.execute(
            select(RuleRelation).where(
                RuleRelation.kind == "co_surfaces",
                or_(
                    RuleRelation.from_rule_id.in_(rule_ids),
                    RuleRelation.to_rule_id.in_(rule_ids),
                ),
            )
        )).scalars().all()
        partners = {
            (edge.to_rule_id if edge.from_rule_id in known else edge.from_rule_id)
            for edge in edges
        } - known
        if not partners:
            return []
        # Ownership re-checked per partner rather than assumed from the edge.
        out = []
        for partner_id in sorted(partners):
            rule = await _fetch_owned_rule(session, partner_id, user_id)
            if rule is not None:
                out.append(rule)
        return out


async def rule_detail(user_id: int, rule: Rule, system_ids: list[int] | None = None) -> dict:
    """The full record, with its areas and edges attached.

    ONE seam for both doors and every write path, so create, update and get
    cannot disagree about what a rule looks like coming back — the same
    reasoning as attach_relations for notes (#2859), and the same reasoning
    rule_brief exists for one level down.

    `system_ids=None` means "leave the tags alone"; a list (including [])
    REPLACES them.
    """
    if system_ids is not None:
        await set_rule_systems(rule.id, user_id, system_ids)
    data = rule.to_dict()
    systems = (await list_rule_systems([rule.id])).get(rule.id, [])
    relations = (await list_rule_relations([rule.id])).get(rule.id, [])
    # Attached only when present (#2483): an empty key reads as a capability
    # the record has and isn't using, which is a different claim.
    if systems:
        data["systems"] = systems
    if relations:
        data["relations"] = relations
    return data


async def create_rule(
    topic_id: int, user_id: int, title: str, statement: str,
    why: str = "", how_to_apply: str = "", order_index: int = 0,
    when_to_apply: str = "", tier: str = "always_on", arose_from_id: int = 0,
    verify_with: str = "", expires_when: str = "", kind: str = "rule",
) -> Rule:
    async with async_session() as session:
        await _assert_topic_owned(session, topic_id, user_id)
        rule = Rule(
            topic_id=topic_id,
            title=title,
            statement=statement,
            when_to_apply=when_to_apply or None,
            tier=_valid_tier(tier),
            kind=_valid_kind(kind),
            why=why or None,
            how_to_apply=how_to_apply or None,
            verify_with=verify_with or None,
            expires_when=expires_when or None,
            arose_from_id=arose_from_id or None,
            order_index=order_index,
        )
        session.add(rule)
        await session.commit()
        await session.refresh(rule)
        _refresh_rule_embedding(rule)
        return rule


async def create_project_rule(
    project_id: int, user_id: int, title: str, statement: str,
    why: str = "", how_to_apply: str = "", order_index: int = 0,
    when_to_apply: str = "", tier: str = "always_on", arose_from_id: int = 0,
    verify_with: str = "", expires_when: str = "", kind: str = "rule",
) -> Rule:
    """Create a rule scoped to a single project (no rulebook ceremony).

    Project-scoped rules apply only to the named project; they don't
    propagate via rulebook subscriptions. Topic_id is left NULL — the
    CHECK constraint enforces exactly-one of (topic_id, project_id).
    """
    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        rule = Rule(
            project_id=project_id,
            title=title,
            statement=statement,
            when_to_apply=when_to_apply or None,
            tier=_valid_tier(tier),
            kind=_valid_kind(kind),
            why=why or None,
            how_to_apply=how_to_apply or None,
            verify_with=verify_with or None,
            expires_when=expires_when or None,
            arose_from_id=arose_from_id or None,
            order_index=order_index,
        )
        session.add(rule)
        await session.commit()
        await session.refresh(rule)
        _refresh_rule_embedding(rule)
        return rule


async def list_rules(
    user_id: int,
    rulebook_id: int | None = None,
    topic_id: int | None = None,
    project_id: int | None = None,
) -> list[Rule]:
    """List rules filtered by any of the three IDs. All filters are ownership-scoped.

    When project_id is set, the result includes both rulebook rules reached via
    project_rulebook_subscriptions AND project-scoped rules (Rule.project_id).
    When rulebook_id or topic_id is set, project-scoped rules are excluded by
    construction (they have neither). With no filter, only rulebook rules are
    returned — adding all of a user's project-scoped rules unprompted would
    surprise existing callers.
    """
    from scribe.models.rulebook import project_rulebook_subscriptions

    async with async_session() as session:
        stmt = (
            select(Rule)
            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .where(
                Rulebook.owner_user_id == user_id,
                Rule.deleted_at.is_(None),
                RulebookTopic.deleted_at.is_(None),
                Rulebook.deleted_at.is_(None),
            )
        )
        if topic_id:
            stmt = stmt.where(Rule.topic_id == topic_id)
        if rulebook_id:
            stmt = stmt.where(RulebookTopic.rulebook_id == rulebook_id)
        if project_id:
            stmt = (
                stmt.join(
                    project_rulebook_subscriptions,
                    project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
                )
                .where(project_rulebook_subscriptions.c.project_id == project_id)
            )
        stmt = stmt.order_by(
            Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
        )
        result = await session.execute(stmt)
        rulebook_rules = list(result.scalars().all())

        if not project_id:
            return rulebook_rules

        # Project-scoped rules (topic_id IS NULL, project_id matches).
        # Verifies ownership by joining Project on user_id.
        from scribe.models.project import Project
        proj_stmt = (
            select(Rule)
            .join(Project, Rule.project_id == Project.id)
            .where(
                Project.user_id == user_id,
                Rule.project_id == project_id,
                Rule.deleted_at.is_(None),
                Project.deleted_at.is_(None),
            )
            .order_by(Rule.order_index, Rule.title)
        )
        proj_result = await session.execute(proj_stmt)
        return rulebook_rules + list(proj_result.scalars().all())


def _excluded_rulebook_ids_q(project_id: int):
    """Subquery: the always-on rulebooks this project opted out of at
    inception (milestone 297) — used by every rule-resolution path so an
    exclusion is total, not just cosmetic."""
    from scribe.models.rulebook import project_rulebook_exclusions

    return select(project_rulebook_exclusions.c.rulebook_id).where(
        project_rulebook_exclusions.c.project_id == project_id
    )


async def excluded_always_on_rulebooks(user_id: int, project_id: int) -> list[dict]:
    """[{id, title}] of the always-on rulebooks excluded for ``project_id``
    (owner-scoped). Empty for an undecided or inherit-all project."""
    from scribe.models.rulebook import project_rulebook_exclusions

    if not project_id:
        return []
    async with async_session() as session:
        rows = (
            await session.execute(
                select(Rulebook.id, Rulebook.title)
                .join(project_rulebook_exclusions,
                      project_rulebook_exclusions.c.rulebook_id == Rulebook.id)
                .where(
                    project_rulebook_exclusions.c.project_id == project_id,
                    Rulebook.owner_user_id == user_id,
                    Rulebook.deleted_at.is_(None),
                )
                .order_by(Rulebook.title)
            )
        ).all()
    return [{"id": rid, "title": title} for rid, title in rows]


async def list_always_on_rules(
    user_id: int, limit: int = 100, project_id: int = 0,
) -> list[Rule]:
    """Return all rules from rulebooks flagged always_on for the user.

    Called by the MCP tool of the same name at session start to load the
    standing rules that apply regardless of which project (if any) is in
    scope. Ordering matches list_rules so results are stable across calls.

    ``project_id`` (milestone 297): inside a project that excluded specific
    always-on rulebooks at inception, those rulebooks' rules are NOT
    returned — the project decided not to inherit them. 0 = the user-wide
    set, which is what a session sees before a project is in scope.
    """
    async with async_session() as session:
        q = (
            select(Rule)
            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .where(
                Rulebook.owner_user_id == user_id,
                Rulebook.always_on.is_(True),
                Rule.deleted_at.is_(None),
                RulebookTopic.deleted_at.is_(None),
                Rulebook.deleted_at.is_(None),
                # TIER (milestone 307). This is the SESSION-START call, made
                # before any project is in scope — there is no area vocabulary
                # to match a conditional rule against yet, so only the
                # unconditional tier belongs here. A conditional rule reaches a
                # session through enter_project (by area) or search (by
                # meaning), not by being resident.
                #
                # Behaviour is unchanged until rules are actually re-tiered:
                # `tier` defaults to always_on, so every existing rule still
                # arrives exactly as it did.
                Rule.tier == "always_on",
            )
        )
        if project_id:
            q = q.where(Rulebook.id.notin_(_excluded_rulebook_ids_q(project_id)))
        result = await session.execute(
            q.order_by(
                Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
            ).limit(limit)
        )
        return list(result.scalars().all())


async def _fetch_owned_rule(session, rule_id: int, user_id: int) -> Optional[Rule]:
    """Fetch a rule by id, scoped to user owning either its rulebook
    (via topic) or its project (via project_id). Honors soft-delete.
    Returns None when not found or not owned.
    """
    from scribe.models.project import Project

    # Path A — rulebook rule.
    rulebook_rule = (await session.execute(
        select(Rule)
        .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
        .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
        .where(
            Rule.id == rule_id,
            Rulebook.owner_user_id == user_id,
            Rule.deleted_at.is_(None),
            RulebookTopic.deleted_at.is_(None),
            Rulebook.deleted_at.is_(None),
        )
    )).scalar_one_or_none()
    if rulebook_rule is not None:
        return rulebook_rule

    # Path B — project-scoped rule.
    project_rule = (await session.execute(
        select(Rule)
        .join(Project, Rule.project_id == Project.id)
        .where(
            Rule.id == rule_id,
            Project.user_id == user_id,
            Rule.deleted_at.is_(None),
            Project.deleted_at.is_(None),
        )
    )).scalar_one_or_none()
    return project_rule


async def get_rule(rule_id: int, user_id: int) -> Optional[Rule]:
    async with async_session() as session:
        return await _fetch_owned_rule(session, rule_id, user_id)


async def update_rule(
    rule_id: int, user_id: int, clear: Iterable[str] = (), **fields,
) -> Optional[Rule]:
    """Patch a rule. `clear` names fields to unset; **fields carries new values.

    Clearing is EXPLICIT and separate because a nullable field cannot be
    emptied by passing it. The MCP door reads "" as "leave this alone" — an
    agent filling three fields must not wipe the other five — so a caller
    there has no value that means "remove it", and a rule that stops being a
    constraint genuinely needs its check removed. Naming the field is the one
    form that cannot happen by accident.

    Callers that DO have a meaningful empty value (the REST door, where a
    cleared form input arrives as "") get the same outcome through
    NULLABLE_RULE_TEXT normalisation below, so the two doors keep their own
    idiom and agree about the result.
    """
    async with async_session() as session:
        rule = await _fetch_owned_rule(session, rule_id, user_id)
        if rule is None:
            return None
        allowed = {
            "title", "statement", "why", "how_to_apply", "order_index",
            "when_to_apply", "tier", "kind", "arose_from_id",
            "verify_with", "expires_when",
        }
        check_before = rule.verify_with
        # Captured BEFORE anything is written, and as plain values — this has
        # to survive the mutation below. A rule's history is the only record
        # of what it used to say; the edit itself destroys that.
        text_before = rule_versions.snapshot(rule)
        for key in clear:
            if key in allowed and key in NULLABLE_RULE_TEXT:
                setattr(rule, key, None)
            elif key == "arose_from_id":
                setattr(rule, key, None)
        for key, value in fields.items():
            if key not in allowed or value is None:
                continue
            if key == "tier":
                value = _valid_tier(value)
            elif key == "kind":
                value = _valid_kind(value)
            elif key in NULLABLE_RULE_TEXT:
                value = value or None
            elif key == "arose_from_id":
                value = value or None
            setattr(rule, key, value)
        # A verification stamp certifies A CHECK, not a rule. Rewrite or
        # remove the check and the old stamp certifies something that no
        # longer exists — so it is dropped, and the rule re-enters the sweep.
        # The safe direction, for the same reason _valid_tier falls back to
        # always_on: a rule wrongly listed as due costs one look, a rule
        # wrongly vouched for costs the thing the sweep exists to catch.
        if rule.verify_with != check_before:
            rule.verified_at = None
        # Same session as the edit, so the two commit together. The snapshot
        # holds the OLD verify_with — the check that was in force when that
        # wording was written — which is why it is taken before the loop and
        # not here.
        rule_versions.record_if_changed(session, rule, user_id, text_before)
        await session.commit()
        await session.refresh(rule)
        _refresh_rule_embedding(rule)
        return rule


# ── Edit history (milestone 323) ───────────────────────────────────────
#
# The ACL-scoped reads live HERE rather than in services/rule_versions.py,
# and not by preference: rulebooks imports rule_versions for the write path,
# so the reverse import would be a cycle. The split is also the honest one —
# rule_versions owns what a version IS, this module owns who may read one.


async def list_rule_versions(rule_id: int, user_id: int):
    """A rule's history, newest first. None when the rule is not readable.

    Scoped through the rule itself, never through the version's `user_id`:
    that column is the ACTOR. Reading a rule's history is a question about
    the RULE, so anyone who can read the rule can read what it used to say,
    and anyone who cannot read the rule gets nothing — including the versions
    they personally wrote, if the rule has since moved out of their reach.
    """
    async with async_session() as session:
        if await _fetch_owned_rule(session, rule_id, user_id) is None:
            return None
    return await rule_versions.list_versions(rule_id)


async def get_rule_version(rule_id: int, version_id: int, user_id: int):
    """One snapshot in full. None when the rule or the version is not found.

    Takes the rule id as well as the version id so the ownership check has
    something to run against BEFORE the version is read, and so a version id
    from another rule cannot be read through a rule the caller does happen to
    own — the check and the fetch have to agree about which rule is in play.
    """
    async with async_session() as session:
        if await _fetch_owned_rule(session, rule_id, user_id) is None:
            return None
        return (await session.execute(
            select(RuleVersion).where(
                RuleVersion.id == version_id,
                RuleVersion.rule_id == rule_id,
            )
        )).scalar_one_or_none()


# ── Canon tags + typed edges (milestone 307) ───────────────────────────

async def set_rule_systems(
    rule_id: int, user_id: int, canonical_ids: list[int],
) -> list[int] | None:
    """Replace which global AREAS a rule is about. None if not owned.

    Set-semantics like set_record_systems: the list given IS the state after,
    so an empty list clears the tags. Points at the canonical catalog, never a
    project's System — a family rule tagged to one project's row would bind
    itself to that project's vocabulary.
    """
    from scribe.models.canonical_system import CanonicalSystem
    from scribe.models.rulebook import rule_systems as rule_systems_t

    async with async_session() as session:
        rule = await _fetch_owned_rule(session, rule_id, user_id)
        if rule is None:
            return None
        wanted = set(canonical_ids or [])
        if wanted:
            live = set((await session.execute(
                select(CanonicalSystem.id).where(
                    CanonicalSystem.id.in_(wanted),
                    CanonicalSystem.deleted_at.is_(None),
                )
            )).scalars().all())
            # Silently dropping an unknown id would leave the caller believing
            # a tag exists; keep only the live ones and report what stuck.
            wanted &= live
        await session.execute(
            sql_delete(rule_systems_t).where(rule_systems_t.c.rule_id == rule_id)
        )
        for canonical_id in sorted(wanted):
            await session.execute(
                insert(rule_systems_t).values(rule_id=rule_id, canonical_id=canonical_id)
            )
        await session.commit()
        return sorted(wanted)


async def list_rule_systems(rule_ids: list[int]) -> dict[int, list[dict]]:
    """The canon tags for a batch of rules, keyed by rule id.

    Batched on purpose: the surfacing paths ask about a whole payload of rules
    at once, and one query per rule would turn every session start into an
    N+1.
    """
    from scribe.models.canonical_system import CanonicalSystem
    from scribe.models.rulebook import rule_systems as rule_systems_t

    if not rule_ids:
        return {}
    async with async_session() as session:
        rows = (await session.execute(
            select(rule_systems_t.c.rule_id, CanonicalSystem.id, CanonicalSystem.name)
            .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
            .where(
                rule_systems_t.c.rule_id.in_(rule_ids),
                CanonicalSystem.deleted_at.is_(None),
            )
            .order_by(CanonicalSystem.order_index)
        )).all()
    out: dict[int, list[dict]] = {}
    for rule_id, canonical_id, name in rows:
        out.setdefault(rule_id, []).append({"id": canonical_id, "name": name})
    return out


async def add_rule_relation(
    user_id: int, from_rule_id: int, to_rule_id: int, kind: str, note: str = "",
) -> RuleRelation | None:
    """Draw a typed edge between two rules. None if either isn't owned.

    Both ends are ownership-checked: an edge is only meaningful if the drawer
    can see both rules, and a one-sided edge would surface a rule the caller
    has no business reading.

    Idempotent — re-drawing an existing edge returns it rather than raising, so
    a true-up pass can be re-run without cleaning up first.
    """
    if kind not in RELATION_KINDS:
        raise ValueError(f"kind must be one of {RELATION_KINDS}, got {kind!r}")
    if from_rule_id == to_rule_id:
        raise ValueError("a rule cannot relate to itself")
    async with async_session() as session:
        for rid in (from_rule_id, to_rule_id):
            if await _fetch_owned_rule(session, rid, user_id) is None:
                return None
        existing = await session.scalar(
            select(RuleRelation).where(
                RuleRelation.from_rule_id == from_rule_id,
                RuleRelation.to_rule_id == to_rule_id,
                RuleRelation.kind == kind,
            )
        )
        if existing is not None:
            return existing
        relation = RuleRelation(
            from_rule_id=from_rule_id, to_rule_id=to_rule_id,
            kind=kind, note=note or None,
        )
        session.add(relation)
        await session.commit()
        await session.refresh(relation)
        return relation


async def remove_rule_relation(user_id: int, relation_id: int) -> bool:
    async with async_session() as session:
        relation = await session.get(RuleRelation, relation_id)
        if relation is None:
            return False
        if await _fetch_owned_rule(session, relation.from_rule_id, user_id) is None:
            return False
        await session.delete(relation)
        await session.commit()
        return True


async def list_rule_relations(rule_ids: list[int]) -> dict[int, list[dict]]:
    """Edges touching a batch of rules, keyed by rule id.

    `co_surfaces` is reported from BOTH ends off a single stored row — it means
    "these fail together", which is not a claim with a direction. The other two
    are directional and are reported as stored, with `direction` naming which
    end this rule is: an override read from the wrong end would invert what it
    says.
    """
    if not rule_ids:
        return {}
    async with async_session() as session:
        rows = (await session.execute(
            select(RuleRelation).where(
                (RuleRelation.from_rule_id.in_(rule_ids))
                | (RuleRelation.to_rule_id.in_(rule_ids))
            )
        )).scalars().all()
    out: dict[int, list[dict]] = {}
    wanted = set(rule_ids)
    for relation in rows:
        if relation.from_rule_id in wanted:
            out.setdefault(relation.from_rule_id, []).append({
                "id": relation.id, "kind": relation.kind,
                "rule_id": relation.to_rule_id,
                "direction": "outgoing", "note": relation.note or "",
            })
        if relation.to_rule_id in wanted:
            out.setdefault(relation.to_rule_id, []).append({
                "id": relation.id, "kind": relation.kind,
                "rule_id": relation.from_rule_id,
                "direction": "incoming", "note": relation.note or "",
            })
    return out


async def delete_rule(rule_id: int, user_id: int) -> None:
    async with async_session() as session:
        rule = await _fetch_owned_rule(session, rule_id, user_id)
        if rule is None:
            return
        await session.delete(rule)
        await session.commit()


# ── Subscriptions + get_applicable_rules ───────────────────────────────

from sqlalchemy.exc import IntegrityError


async def subscribe_project(
    project_id: int, rulebook_id: int, user_id: int,
) -> None:
    """Add a subscription. Idempotent — duplicates raise; we swallow."""
    from scribe.models.rulebook import project_rulebook_subscriptions

    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await _assert_rulebook_owned(session, rulebook_id, user_id)
        # ON CONFLICT DO NOTHING via try/except to keep dialect-agnostic.
        try:
            await session.execute(
                insert(project_rulebook_subscriptions).values(
                    project_id=project_id, rulebook_id=rulebook_id,
                )
            )
            await session.commit()
        except Exception:
            await session.rollback()  # PK collision = already subscribed; fine.


async def unsubscribe_project(
    project_id: int, rulebook_id: int, user_id: int,
) -> None:
    from scribe.models.rulebook import project_rulebook_subscriptions

    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await _assert_rulebook_owned(session, rulebook_id, user_id)
        await session.execute(
            sql_delete(project_rulebook_subscriptions).where(
                project_rulebook_subscriptions.c.project_id == project_id,
                project_rulebook_subscriptions.c.rulebook_id == rulebook_id,
            )
        )
        await session.commit()


# ── Suppressions — project-level mute of rulebook rules / topics ────────

async def suppress_rule_for_project(
    project_id: int, rule_id: int, user_id: int,
) -> None:
    """Mute one rulebook rule for one project. Idempotent."""
    from scribe.models.rulebook import project_rule_suppressions

    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await _assert_rulebook_rule_owned(session, rule_id, user_id)
        try:
            await session.execute(
                insert(project_rule_suppressions).values(
                    project_id=project_id, rule_id=rule_id,
                )
            )
            await session.commit()
        except Exception:
            await session.rollback()  # PK collision = already suppressed; fine.


async def unsuppress_rule_for_project(
    project_id: int, rule_id: int, user_id: int,
) -> None:
    """Unmute one rulebook rule for one project. Idempotent."""
    from scribe.models.rulebook import project_rule_suppressions

    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await session.execute(
            sql_delete(project_rule_suppressions).where(
                project_rule_suppressions.c.project_id == project_id,
                project_rule_suppressions.c.rule_id == rule_id,
            )
        )
        await session.commit()


async def exclude_always_on_rulebook_for_project(
    project_id: int, rulebook_id: int, user_id: int,
) -> None:
    """Opt one project out of a whole ALWAYS-ON rulebook (milestone 297).
    Owner-only on both sides; the rulebook must be always_on — a subscribed
    rulebook is left by unsubscribing, not excluding. Idempotent."""
    from scribe.models.rulebook import project_rulebook_exclusions

    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await _assert_rulebook_owned(session, rulebook_id, user_id)
        rb = await session.get(Rulebook, rulebook_id)
        if rb is None or not rb.always_on:
            raise ValueError(
                f"rulebook {rulebook_id} is not always-on — it binds only by "
                "subscription; unsubscribe_project_from_rulebook instead"
            )
        try:
            await session.execute(
                insert(project_rulebook_exclusions).values(
                    project_id=project_id, rulebook_id=rulebook_id,
                )
            )
            await session.commit()
        except IntegrityError:
            await session.rollback()  # already excluded — idempotent


async def include_always_on_rulebook_for_project(
    project_id: int, rulebook_id: int, user_id: int,
) -> None:
    """Undo exclude_always_on_rulebook_for_project. Idempotent."""
    from scribe.models.rulebook import project_rulebook_exclusions

    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await session.execute(
            sql_delete(project_rulebook_exclusions).where(
                project_rulebook_exclusions.c.project_id == project_id,
                project_rulebook_exclusions.c.rulebook_id == rulebook_id,
            )
        )
        await session.commit()


async def suppress_topic_for_project(
    project_id: int, topic_id: int, user_id: int,
) -> None:
    """Mute every rule under one topic for one project. Idempotent."""
    from scribe.models.rulebook import project_topic_suppressions

    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await _assert_topic_owned(session, topic_id, user_id)
        try:
            await session.execute(
                insert(project_topic_suppressions).values(
                    project_id=project_id, topic_id=topic_id,
                )
            )
            await session.commit()
        except Exception:
            await session.rollback()


async def unsuppress_topic_for_project(
    project_id: int, topic_id: int, user_id: int,
) -> None:
    """Unmute a topic for one project. Idempotent."""
    from scribe.models.rulebook import project_topic_suppressions

    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await session.execute(
            sql_delete(project_topic_suppressions).where(
                project_topic_suppressions.c.project_id == project_id,
                project_topic_suppressions.c.topic_id == topic_id,
            )
        )
        await session.commit()


async def get_applicable_rules(
    project_id: int, user_id: int, limit: int = 50,
) -> dict:
    """Return rules applicable to a project — both via rulebook subscriptions
    and project-scoped rules (Rule.project_id matches), with suppressed rules
    and suppressed topics filtered out.

    Shape:
        {
          "rules": [{id, title, statement,
                     topic_id, topic_title,
                     rulebook_id, rulebook_title}, ...],
          "project_rules": [{id, title, statement}, ...],
          "suppressed_rules": [{id, title,
                                topic_id, topic_title,
                                rulebook_id, rulebook_title}, ...],
          "suppressed_topics": [{id, title,
                                 rulebook_id, rulebook_title}, ...],
          "truncated": bool,
          "subscribed_rulebooks": [{id, title}, ...]
        }

    `rules` is the subscription-derived set with project-level suppressions
    applied. `project_rules` is the project-scoped set (never suppressed —
    delete instead). `suppressed_rules` / `suppressed_topics` carry the
    titles + rulebook context callers need to display what was filtered
    without round-tripping for names.
    """
    from scribe.models.rulebook import (
        project_rulebook_subscriptions,
        project_rule_suppressions,
        project_topic_suppressions,
    )

    async with async_session() as session:
        # Subscribed rulebooks for the project (ownership-scoped).
        sub_q = (
            select(Rulebook.id, Rulebook.title)
            .join(
                project_rulebook_subscriptions,
                project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
            )
            .where(
                project_rulebook_subscriptions.c.project_id == project_id,
                Rulebook.owner_user_id == user_id,
                Rulebook.deleted_at.is_(None),
            )
            .order_by(Rulebook.title)
        )
        sub_rows = (await session.execute(sub_q)).all()
        subscribed_rulebooks = [
            {"id": rb_id, "title": rb_title} for rb_id, rb_title in sub_rows
        ]

        # Suppressed rules — joined to topic + rulebook so callers can render
        # context without a follow-up lookup. Ownership-scoped via rulebook.
        suppressed_rules_q = (
            select(
                Rule.id, Rule.title,
                RulebookTopic.id.label("topic_id"),
                RulebookTopic.title.label("topic_title"),
                Rulebook.id.label("rulebook_id"),
                Rulebook.title.label("rulebook_title"),
            )
            .join(project_rule_suppressions, project_rule_suppressions.c.rule_id == Rule.id)
            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .where(
                project_rule_suppressions.c.project_id == project_id,
                Rulebook.owner_user_id == user_id,
            )
            .order_by(Rulebook.title, RulebookTopic.title, Rule.title)
        )
        suppressed_rule_rows = (await session.execute(suppressed_rules_q)).all()
        suppressed_rules = [
            {"id": rid, "title": rt, "topic_id": ti, "topic_title": tt,
             "rulebook_id": rbi, "rulebook_title": rbt}
            for rid, rt, ti, tt, rbi, rbt in suppressed_rule_rows
        ]
        suppressed_rule_ids = [r["id"] for r in suppressed_rules]

        # Suppressed topics — joined to rulebook for context.
        suppressed_topics_q = (
            select(
                RulebookTopic.id, RulebookTopic.title,
                Rulebook.id.label("rulebook_id"),
                Rulebook.title.label("rulebook_title"),
            )
            .join(project_topic_suppressions, project_topic_suppressions.c.topic_id == RulebookTopic.id)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .where(
                project_topic_suppressions.c.project_id == project_id,
                Rulebook.owner_user_id == user_id,
            )
            .order_by(Rulebook.title, RulebookTopic.title)
        )
        suppressed_topic_rows = (await session.execute(suppressed_topics_q)).all()
        suppressed_topics = [
            {"id": tid, "title": tt, "rulebook_id": rbi, "rulebook_title": rbt}
            for tid, tt, rbi, rbt in suppressed_topic_rows
        ]
        suppressed_topic_ids = [t["id"] for t in suppressed_topics]

        # Applicable rules (limit + 1 so we can detect truncation). Filter
        # in SQL so truncation reflects the post-suppression count, not the
        # raw subscription count.
        # Selects the ENTITY, not a column list: rule_brief is the one place
        # that decides which fields a surfaced rule carries, and a column list
        # here would be a second such decision to keep in step. The row count
        # is bounded by `limit`, so this is a listing, not a scan.
        rules_q = (
            select(
                Rule,
                RulebookTopic.title.label("topic_title"),
                Rulebook.id.label("rulebook_id"),
                Rulebook.title.label("rulebook_title"),
            )
            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .join(
                project_rulebook_subscriptions,
                project_rulebook_subscriptions.c.rulebook_id == Rulebook.id,
            )
            .where(
                project_rulebook_subscriptions.c.project_id == project_id,
                Rulebook.owner_user_id == user_id,
                Rule.deleted_at.is_(None),
                RulebookTopic.deleted_at.is_(None),
                Rulebook.deleted_at.is_(None),
                # An inception exclusion is total (milestone 297): a rulebook the
                # project opted out of contributes nothing, subscribed or not.
                Rulebook.id.notin_(_excluded_rulebook_ids_q(project_id)),
            )
            .order_by(
                Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
            )
            .limit(limit + 1)
        )
        if suppressed_rule_ids:
            rules_q = rules_q.where(Rule.id.notin_(suppressed_rule_ids))
        if suppressed_topic_ids:
            rules_q = rules_q.where(Rule.topic_id.notin_(suppressed_topic_ids))
        # TIER (milestone 307). always_on rules are resident, as every rule was
        # before tiers existed. A conditional rule is REACHABLE, and reaches
        # this project only when it is tagged to an area this project actually
        # works in — a deterministic tag match, never a similarity score, so
        # bindingness never depends on a ranking (D7).
        #
        # Applied in SQL rather than by filtering afterwards, so `limit` counts
        # the rules that will actually be surfaced instead of counting rules
        # that are about to be dropped.
        project_area_ids = (await session.execute(
            select(System.canonical_id).where(
                System.project_id == project_id,
                System.canonical_id.is_not(None),
                System.deleted_at.is_(None),
                System.status == "active",
            ).distinct()
        )).scalars().all()
        reachable = select(rule_systems.c.rule_id).where(
            rule_systems.c.canonical_id.in_(project_area_ids)
        ) if project_area_ids else None
        tier_clause = (Rule.tier == "always_on")
        if reachable is not None:
            tier_clause = or_(tier_clause, Rule.id.in_(reachable))
        rules_q = rules_q.where(tier_clause)
        rule_rows = (await session.execute(rules_q)).all()
        truncated = len(rule_rows) > limit
        rules = [
            rule_brief(rule, topic_title=tt, rulebook_id=rbi, rulebook_title=rbt)
            for rule, tt, rbi, rbt in rule_rows[:limit]
        ]

        # Project-scoped rules — verifies ownership via Project.user_id.
        from scribe.models.project import Project
        proj_rules_q = (
            select(Rule)
            .join(Project, Rule.project_id == Project.id)
            .where(
                Project.user_id == user_id,
                Rule.project_id == project_id,
                Rule.deleted_at.is_(None),
                Project.deleted_at.is_(None),
            )
            .order_by(Rule.order_index, Rule.title)
        )
        if reachable is not None:
            proj_rules_q = proj_rules_q.where(
                or_(Rule.tier == "always_on", Rule.id.in_(reachable))
            )
        else:
            proj_rules_q = proj_rules_q.where(Rule.tier == "always_on")
        proj_rule_rows = (await session.execute(proj_rules_q)).all()
        project_rules = [rule_brief(rule) for (rule,) in proj_rule_rows]

    # Edges travel with the rules they belong to (milestone 307).
    #
    # A co_surfaces partner that was not otherwise selected is ADDED, because a
    # rule that arrives without the half it fails with is the failure the edge
    # was created to prevent. Suppressions are passed as exclusions so an
    # explicit mute still wins over an edge.
    surfaced_ids = [r["id"] for r in rules] + [r["id"] for r in project_rules]
    partners = await co_surfaced_partners(
        user_id, surfaced_ids, exclude_ids=set(suppressed_rule_ids),
    )
    for partner in partners:
        rules.append(rule_brief(partner, via="co_surfaces"))
        surfaced_ids.append(partner.id)

    # Relations on every surfaced rule, so a reader can see that an override
    # exists rather than discovering the contradiction by acting on the wrong
    # one. Areas too — they are why a conditional rule is here at all.
    edges = await list_rule_relations(surfaced_ids)
    areas = await list_rule_systems(surfaced_ids)
    for brief in (*rules, *project_rules):
        if edges.get(brief["id"]):
            brief["relations"] = edges[brief["id"]]
        if areas.get(brief["id"]):
            brief["systems"] = areas[brief["id"]]

    return {
        "rules": rules,
        "project_rules": project_rules,
        "suppressed_rules": suppressed_rules,
        "suppressed_topics": suppressed_topics,
        "truncated": truncated,
        "subscribed_rulebooks": subscribed_rulebooks,
        "excluded_always_on": await excluded_always_on_rulebooks(user_id, project_id),
    }


def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict:
    """The caller-facing shape of a get_applicable_rules() result.

    Every surface that hands rules to an agent (enter_project, get_project,
    get_milestone, get_task for legacy plans, start_planning) carries the
    same seven keys under the same names — so a reader learns them once. One
    place renames `rules` → `applicable_rules` and `truncated` →
    `applicable_rules_truncated`; the tools merge this into their payloads.
    `excluded_always_on` (milestone 297) names the always-on rulebooks this
    project decided NOT to inherit, so the departure is visible wherever the
    rules are.

    IT ALSO RECORDS THE SURFACING, which is why it now takes a caller and a
    source. Every one of those surfaces is a bulk delivery — the applicable set
    handed over whole, chosen by nobody — so this is the one place that has to
    emit for all of them. Doing it per-caller instead would be five sites to
    remember, and #3430 gap 2 is what that costs: the process→skill sync went
    un-emitted through an entire dedicated telemetry survey because nothing
    forced its surface to be accounted for.

    `source` stays the CALLER's name rather than a constant, so the readout can
    still separate the session handshake from a mid-session milestone read;
    `RANKED_SOURCES` in `rule_usage` is what folds them back together.

    Emitting from here is safe in a way emitting from `get_applicable_rules`
    would not be: this function is only ever called to BUILD A REPLY. The two
    other callers of the rules machinery — the write-path etag arm
    (`plugin_context`) and `rules_etag_for` — compute a marker and show nobody
    anything, and counting those would put rules in the denominator that no
    agent ever saw.
    """
    record_rule_surfaced(
        user_id=user_id,
        rule_ids=(
            [r["id"] for r in applicable.get("rules", [])]
            + [r["id"] for r in applicable.get("project_rules", [])]
        ),
        source=source,
    )
    return {
        "applicable_rules": applicable["rules"],
        "applicable_rules_truncated": applicable["truncated"],
        "subscribed_rulebooks": applicable["subscribed_rulebooks"],
        "project_rules": applicable.get("project_rules", []),
        "suppressed_rules": applicable.get("suppressed_rules", []),
        "suppressed_topics": applicable.get("suppressed_topics", []),
        "excluded_always_on": applicable.get("excluded_always_on", []),
    }


# ── The staleness marker (milestone 323 step 5) ────────────────────────
#
# WHAT THIS CAN AND CANNOT SEE. An etag catches a rule that MOVED after a
# session loaded it. It is not a general staleness check, and a reader who
# finds one here will assume it is:
#
#   what goes wrong                                    | caught?
#   ---------------------------------------------------|--------
#   another session edits a rule mid-flight             | yes
#   the session is misremembering a rule read hours ago | yes
#   compaction summarised the rules out of context      | NO
#
# The third is the most common and the marker is blind to it, because the
# etag was in context too and went with the rules. The SessionStart nudge is
# that case's only mechanism, and MUST NOT be softened because this exists —
# retiring something that covers the common case in favour of something that
# does not is the plausible mistake here.

_ETAG_EMPTY = "empty|0"


def rules_etag(rules: list) -> str:
    """A marker for "is the set you are holding still the current one?".

    `max(updated_at)` alone is not enough: DELETING a rule moves no timestamp,
    and that is the single change that takes an instruction OUT of force —
    the one a session most needs to hear about. The count catches it.

    Instance-agnostic (rule 115): it knows nothing about any particular
    rulebook, and an install with one rule or none produces a stable marker
    rather than an error. "No rules" must read as a state, not as a change,
    or every session on a fresh install would be told its rules had moved.
    """
    if not rules:
        return _ETAG_EMPTY
    # A decoration must not be able to break what it decorates. This is
    # computed on the SessionStart path, where raising would cost the whole
    # context payload to save a hint — so a row with no usable timestamp is
    # skipped rather than compared, and a set with none degrades to a
    # count-only marker instead of failing. Count-only still catches a rule
    # added or deleted; it just cannot see an edit, which is the right way
    # round to lose information.
    stamps = [
        r.updated_at for r in rules
        if isinstance(getattr(r, "updated_at", None), datetime)
    ]
    if not stamps:
        return f"unknown|{len(rules)}"
    return f"{max(stamps).isoformat()}|{len(rules)}"


async def rules_etag_for(user_id: int, project_id: int = 0) -> str:
    """The current marker for the set a session at this scope would hold.

    Deliberately built from `list_always_on_rules` rather than from a
    `max()/count()` aggregate. An aggregate would be cheaper, and would have
    to restate that function's definition of the set — the always_on flag,
    the project's inception exclusions, the tier filter. Two definitions of
    "the session's rules" is how the marker starts disagreeing with the
    rules, which is worse than materialising a few dozen rows.
    """
    rules = await list_always_on_rules(user_id, project_id=project_id)
    return rules_etag(rules)


def rules_moved_since(rules: list, held_etag: str) -> list:
    """The rules whose text changed after `held_etag` was issued.

    Returns [] when the marker matches, is unparseable, or is absent — a
    caller cannot act on "something is different but I cannot say what", and
    a garbled marker must not be reported as a change.

    A count difference is real news that this list cannot show: a rule
    DELETED since the marker was issued has no row left to return. Callers
    compare counts separately.
    """
    if not held_etag or held_etag == _ETAG_EMPTY:
        return []
    stamp, _, _count = held_etag.partition("|")
    try:
        held_at = datetime.fromisoformat(stamp)
    except ValueError:
        return []
    return [r for r in rules if r.updated_at and r.updated_at > held_at]


def etag_count(held_etag: str) -> int | None:
    """How many rules the holder had. None when the marker cannot be read."""
    _stamp, _, count = (held_etag or "").partition("|")
    try:
        return int(count)
    except ValueError:
        return None


# ── The staleness sweep (milestone 312) ────────────────────────────────

async def rules_due_for_verification(
    user_id: int,
    older_than_days: int = 0,
    tier: str = "",
    never_only: bool = False,
) -> list[Rule]:
    """Rules that carry a check, oldest verification first, never-checked top.

    THE QUERY THIS MILESTONE EXISTS FOR. `verify_with` and `expires_when` are
    storage; this is what turns them into something that gets acted on. The
    307 audit cost a session and found four broken rules by luck — this makes
    the same question a list, and staleness measurable by age instead of
    discoverable by accident.

    Ordered `verified_at` ASC NULLS FIRST: never-checked outranks
    checked-long-ago, because a rule nobody has ever confirmed is a claim
    with no evidence behind it at all.

    Rules with no `verify_with` never appear. That is not an omission — they
    are decisions, there is nothing to go and check, and listing them would
    dilute the result until nobody reads it.

    Ownership-scoped exactly like list_rules: a rule reached through an owned
    rulebook, or scoped to an owned project. Rules have no sharing ACL in this
    schema — no rule_shares, no rulebook_shares — so there is no wider set to
    consult here, unlike notes and projects.

    Args:
        user_id: whose rules.
        older_than_days: only rules last verified longer ago than this.
            Never-checked rules always qualify — they are the most overdue
            thing there is. 0 = no age filter.
        tier: "always_on" or "conditional" to narrow. Raises on anything else
            rather than falling back: _valid_tier's silent always_on default
            is right for a WRITE (the safe direction is to keep binding), and
            wrong for a FILTER, where it would quietly answer a different
            question than the one asked.
        never_only: only rules that have never been verified.
    """
    from datetime import datetime, timedelta, timezone

    from scribe.models.project import Project

    if tier and tier not in TIERS:
        raise ValueError(f"tier must be one of {TIERS}, got {tier!r}")

    async with async_session() as session:
        stmt = (
            select(Rule)
            .outerjoin(RulebookTopic, Rule.topic_id == RulebookTopic.id)
            .outerjoin(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .outerjoin(Project, Rule.project_id == Project.id)
            .where(
                Rule.deleted_at.is_(None),
                Rule.verify_with.is_not(None),
                # One statement rather than two queries merged in Python, so
                # the ordering below is the database's and cannot disagree
                # with itself across the two halves of the XOR.
                or_(
                    and_(
                        Rulebook.owner_user_id == user_id,
                        Rulebook.deleted_at.is_(None),
                        RulebookTopic.deleted_at.is_(None),
                    ),
                    Project.user_id == user_id,
                ),
            )
        )
        if tier:
            stmt = stmt.where(Rule.tier == tier)
        if never_only:
            stmt = stmt.where(Rule.verified_at.is_(None))
        elif older_than_days > 0:
            cutoff = datetime.now(timezone.utc) - timedelta(days=older_than_days)
            stmt = stmt.where(
                or_(Rule.verified_at.is_(None), Rule.verified_at < cutoff)
            )
        stmt = stmt.order_by(Rule.verified_at.asc().nullsfirst(), Rule.id)
        return list((await session.execute(stmt)).scalars().all())


def verification_row(rule: Rule) -> dict:
    """One row of the sweep — the CHECK in full, unlike rule_brief.

    The opposite call from a listing: here the caller is about to go and run
    the check, so the text they need is the point of the payload rather than
    the bloat. `days_since` is computed rather than left to the reader,
    because "2026-06-14" and "74 days" prompt different reactions and only
    one of them is the question being asked.
    """
    days = _days_since_verified(rule)
    return {
        "id": rule.id,
        "title": rule.title,
        "statement": rule.statement,
        "tier": rule.tier,
        "topic_id": rule.topic_id,
        "project_id": rule.project_id,
        "when_to_apply": rule.when_to_apply or "",
        "verify_with": rule.verify_with or "",
        "expires_when": rule.expires_when or "",
        "last_verified": last_verified_label(rule),
        "days_since_verified": days,
    }


async def mark_rule_verified(
    rule_id: int, user_id: int, still_true: bool = True,
) -> Optional[Rule]:
    """Stamp a rule as verified — or, when the check FAILED, refuse to.

    A failing check is the outcome worth having, and the asymmetry is
    deliberate: passing writes a stamp, failing writes nothing. There is no
    "verified false" state to record, because a rule whose check failed is
    not a rule in a special condition — it is a rule that is WRONG, and the
    only honest resolutions are to correct it, retire it, or find out why.
    Recording the failure as a flag would let it sit there being false with
    the sweep quietly satisfied that someone had looked.

    So a failed check leaves `verified_at` untouched, and the rule stays at
    the top of the sweep until someone actually deals with it.

    Returns None when the rule is not found, not owned, or carries no
    `verify_with` — nothing to verify is a different answer from verified.
    """
    from datetime import datetime, timezone

    async with async_session() as session:
        rule = await _fetch_owned_rule(session, rule_id, user_id)
        if rule is None or not rule.verify_with:
            return None
        if still_true:
            rule.verified_at = datetime.now(timezone.utc)
            await session.commit()
            await session.refresh(rule)
        return rule