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 typing import Optional

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

from scribe.models import async_session
from scribe.models.base import iso
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"}
        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 its topics and rules."""
    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")


# 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).
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_kind(kind: str) -> str:
    """An unrecognised kind falls back to `rule` — the SAFE direction.

    The unrecognised value falls back to the binding one — the SAFE
    direction, 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,
        # 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]) -> 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.

    Only rules the caller OWNS: an edge is not a back door into someone else's
    rulebook. Whether a partner can reach a given PROJECT is the caller's
    question (get_applicable_rules drops another project's rule), because this
    answers "what fails with these", which has no project in it.
    """
    if not rule_ids:
        return []
    known = set(rule_ids)
    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,
    moments: list[str] | None = None,
) -> dict:
    """The full record, with its areas, moments 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` / `moments=None` mean "leave them alone"; a list
    (including []) REPLACES them. Doors validate `moments` with
    `moments.require_moments` BEFORE their create, so an unknown name is
    refused without leaving a rule behind.
    """
    if system_ids is not None:
        await set_rule_systems(rule.id, user_id, system_ids)
    if moments is not None:
        await set_rule_moments(rule.id, user_id, moments)
    data = rule.to_dict()
    systems = (await list_rule_systems([rule.id])).get(rule.id, [])
    mounted = (await list_rule_moments([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 mounted:
        data["moments"] = mounted
    if relations:
        data["relations"] = relations
    # The concrete situations judged (or proposed) to be instances of this
    # rule — milestone 440. Same present-only convention as the two above.
    from scribe.services.lesson_rules import attach_rule_lessons

    await attach_rule_lessons(user_id, data, rule.id)
    return data


def _require_trigger(when_to_apply: str | None) -> None:
    """A rule without a trigger is not a quiet rule — it is an unreachable one.

    Nothing is preloaded, so `when_to_apply` is the whole of how a rule
    arrives. It is also what the record is EMBEDDED as: `rule_document` builds
    `{title} — {trigger}` / `When to apply: {trigger}\\n\\n{statement}`, with
    the trigger appearing twice so that purpose dominates a short vector. Drop
    it and the document silently changes shape to title + statement, so the
    same score means something different for that rule than for its
    neighbours — and every bar and every rank in the system assumes one shape.

    ENFORCED IN THE SERVICE, so both doors are covered: the MCP tools and the
    frontend's fast path (`routes/rulebooks.py`) both land here, and a guard
    written in one of them would leave the other able to create a rule that
    never fires.

    Deliberately NOT following `arose_from_id`, which the human door exempts
    itself from on the stated grounds that provenance is about auditing what
    the AGENT changed. That reasoning does not reach this field: a missing
    trigger is not a missing explanation, it is a rule that does not work, and
    it fails an operator exactly as badly as it fails a session.
    """
    if not (when_to_apply or "").strip():
        raise ValueError(
            "when_to_apply is required: a rule with no trigger never surfaces "
            "at the moment it applies. Name that moment in the words a session "
            "would actually be producing then — the command, the error, the "
            "half-formed ask — not the category it belongs to."
        )


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 = "", arose_from_id: int = 0,
    verify_with: str = "", expires_when: str = "", kind: str = "rule",
) -> Rule:
    _require_trigger(when_to_apply)
    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,
            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 = "", 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: retrieval surfaces
    them in that project's sessions and nowhere else (milestone 414), where a
    rule in a rulebook topic is global. Topic_id is left NULL — the CHECK
    constraint enforces exactly-one of (topic_id, project_id).
    """
    _require_trigger(when_to_apply)
    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,
            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 by rulebook, topic or project. Ownership-scoped.

    A rule has one home (milestone 414): a rulebook topic, where it is global,
    or a project. So the filters name homes rather than reach:

    - `project_id` lists that project's OWN rules. Global rules apply to every
      project, so listing them under each one would say nothing; list them by
      rulebook, or unfiltered. `rulebook_id` / `topic_id` don't combine with it
      — a project rule has neither.
    - `rulebook_id` / `topic_id` list global rules in that rulebook or topic.
    - No filter lists every global rule. A user's project rules are left out:
      they belong to their projects, and mixing them into the rulebook listing
      would surprise its callers.

    Before milestone 414, `project_id` returned the rules of every rulebook the
    project SUBSCRIBED to plus its own. Subscriptions are gone.
    """
    from scribe.models.project import Project

    async with async_session() as session:
        if project_id:
            result = await session.execute(
                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)
            )
            return list(result.scalars().all())

        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)
        stmt = stmt.order_by(
            Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
        )
        result = await session.execute(stmt)
        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", "kind", "arose_from_id",
            "verify_with", "expires_when",
        }
        check_before = rule.verify_with
        # A create-time guard is worth nothing if an edit can undo it, and
        # both doors can: `clear=["when_to_apply"]` from the MCP side, and a
        # emptied form input normalised to None from the REST side. Checked
        # AFTER the mutation instead, so it covers every route to an empty
        # trigger including ones added later.
        #
        # Asked as "did this edit REMOVE a trigger", not "does one exist":
        # a rule predating the guard has none, and refusing to save it would
        # make the record permanently unfixable — freezing the exact rules
        # that most need the edit.
        trigger_before = (rule.when_to_apply or "").strip()
        # 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 == "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: 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
        if trigger_before and not (rule.when_to_apply or "").strip():
            raise ValueError(
                "when_to_apply cannot be cleared: it is how this rule arrives, "
                "and it is half of what the rule is embedded as. Replace the "
                "trigger with a better one rather than removing it — a rule "
                "with none is not a quieter rule, it is an unreachable one."
            )
        # 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


async def move_rule(
    rule_id: int, user_id: int, *, topic_id: int = 0, project_id: int = 0,
) -> Optional[Rule]:
    """Give a rule a new home — into a rulebook topic (global) or onto a
    project — keeping its id, history, Systems and relations (milestone 414).

    A rule's home IS its reach: in a topic it applies to every project, on a
    project to that project alone. Recreating the rule in the other home and
    trashing the original would lose its id (and every record citing it), its
    edit history, its area tags and its typed edges, which is why this exists.

    Exactly one of `topic_id` / `project_id`, matching the model's CHECK
    (migration 0059). Raises ValueError for: neither or both named, a target
    the caller does not own, the rule already living there, or a topic that
    already holds a live rule with this title (uq_rule_per_topic) — the message
    names that rule, rather than letting the constraint fail the commit.
    Returns None when the rule itself is not the caller's.

    WHAT A MOVE DOES NOT DO, deliberately:

    - No version. A rule's history records its TEXT (milestone 323, decision
      4); its place is not text, and folding it in would make "version" mean
      two things. The rule's `updated_at` moves; say why a rule moved where
      the decision is recorded.
    - No duplicate gate. Nothing new enters the corpus — the same rule changes
      home — so there is no second record to warn about.
    - No re-embed. The rule's document is its title, statement and trigger;
      retrieval reads the home from the row at query time.
    """
    if bool(topic_id) == bool(project_id):
        raise ValueError("name exactly one destination: topic_id (global) or project_id")
    async with async_session() as session:
        rule = await _fetch_owned_rule(session, rule_id, user_id)
        if rule is None:
            return None
        if topic_id:
            if rule.topic_id == topic_id:
                raise ValueError(f"rule {rule_id} is already in topic {topic_id}")
            await _assert_topic_owned(session, topic_id, user_id)
            clash = (await session.execute(
                select(Rule.id).where(
                    Rule.topic_id == topic_id,
                    Rule.title == rule.title,
                    Rule.deleted_at.is_(None),
                    Rule.id != rule.id,
                )
            )).scalar_one_or_none()
            if clash is not None:
                raise ValueError(
                    f'topic {topic_id} already has a rule titled "{rule.title}" '
                    f"(rule {clash}) — rename one before moving"
                )
            rule.project_id = None
            rule.topic_id = topic_id
        else:
            if rule.project_id == project_id:
                raise ValueError(f"rule {rule_id} is already on project {project_id}")
            await _assert_project_owned(session, project_id, user_id)
            rule.topic_id = None
            rule.project_id = project_id
        await session.commit()
        await session.refresh(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()


# ── Drift: what Scribe changed about how it works with you (#3895) ─────
#
# A preference is the one record kind the AGENT rewrites in the ordinary
# course of working, which is the property that makes it useful and the
# property that makes it dangerous. Milestone 399 named the risk up front:
# an agent misreads one session, rewrites a preference, and follows the
# rewritten version forever while the operator never sees the moment it
# changed — a confident wrong answer wearing the operator's own authority.
#
# `rule_versions` already records every rewrite. What it does not do is
# ARRIVE: a history you have to open one rule at a time, having first
# suspected that rule, is not oversight. These two functions are the push
# half — one read that answers "what changed lately", and one write that
# puts it back.

_DRIFT_LIMIT_MAX = 50


def _owned_rules_clause(user_id: int):
    """Rules whose rulebook or project this user owns — the LISTING form of
    `_fetch_owned_rule`.

    Deliberately the same two paths in the same order, because a listing that
    admits a rule the per-row fetch would refuse is a leak, and one that
    refuses a rule the fetch admits is a row the operator cannot act on. The
    per-row version stays the authority: everything reached through this is
    re-checked by `_fetch_owned_rule` before it is written to.
    """
    from scribe.models.project import Project

    via_rulebook = (
        select(RulebookTopic.id)
        .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
        .where(
            Rulebook.owner_user_id == user_id,
            RulebookTopic.deleted_at.is_(None),
            Rulebook.deleted_at.is_(None),
        )
    )
    via_project = select(Project.id).where(
        Project.user_id == user_id, Project.deleted_at.is_(None),
    )
    return and_(
        Rule.deleted_at.is_(None),
        or_(Rule.topic_id.in_(via_rulebook), Rule.project_id.in_(via_project)),
    )


async def recent_preference_drift(user_id: int, limit: int = 20) -> list[dict]:
    """Preferences that have been rewritten, most recently changed first.

    ONE ROW PER PREFERENCE, carrying its LATEST rewrite — not every version of
    every preference. The question this answers is "what has Scribe changed
    about how it works with me lately", and a preference rewritten eight times
    this week is one answer to that, not eight. The full history of any one
    preference is still a click away in its editor, which is where "how did
    this get here" belongs.

    Each row carries what it said before, what it says now, and `taught_by` —
    the record named by `arose_from_id`, which step 2 made required on every
    agent-written preference. That is the provenance: not who typed it (the
    agent acts as the operator's own user, so the actor column cannot tell
    them apart) but what the change was learned from.

    Rules are excluded, and not as a filter that could be relaxed. A rule
    changes when its author changes it, so "what changed without me" is not a
    question about rules — including them would bury the few rows that are
    actually unreviewed under every edit the operator made themselves.
    """
    limit = max(1, min(int(limit), _DRIFT_LIMIT_MAX))
    # The newest version per rule, by id rather than by created_at: a version
    # is written once and never updated, so id order IS time order, and two
    # versions written in the same clock tick still have a defined winner.
    newest = (
        select(
            RuleVersion.rule_id.label("rule_id"),
            func.max(RuleVersion.id).label("version_id"),
        )
        .group_by(RuleVersion.rule_id)
        .subquery()
    )
    async with async_session() as session:
        rows = (await session.execute(
            select(Rule, RuleVersion)
            .join(newest, newest.c.rule_id == Rule.id)
            .join(RuleVersion, RuleVersion.id == newest.c.version_id)
            .where(Rule.kind == "preference", _owned_rules_clause(user_id))
            .order_by(RuleVersion.created_at.desc(), RuleVersion.id.desc())
            .limit(limit)
        )).all()

        # One query for every provenance record, not one per row. These are
        # titles for rows the caller can already see, so they are read without
        # a further ownership filter — the same reasoning as
        # milestones.titles_for, and blanking them would leave the operator a
        # bare id where the whole point is naming the record.
        taught_ids = {r.arose_from_id for r, _v in rows if r.arose_from_id}
        titles: dict[int, str] = {}
        if taught_ids:
            from scribe.models.note import Note

            titles = dict((await session.execute(
                select(Note.id, Note.title).where(
                    Note.id.in_(taught_ids), Note.deleted_at.is_(None),
                )
            )).all())

    out: list[dict] = []
    for rule, version in rows:
        row: dict = {
            # rule_brief, not a hand-written dict: the drift pane links
            # straight into the editor, so the row has to be the same shape
            # every other rule listing is (#3313's three-copies lesson).
            "rule": rule_brief(rule),
            # What it said BEFORE this rewrite. `statement` and
            # `when_to_apply` only — those are the two fields that change how
            # a session behaves, and a diff of `why` is reading rather than
            # reviewing.
            "previous": {
                "id": version.id,
                "created_at": iso(version.created_at),
                "title": version.title or "",
                "statement": version.statement or "",
                "when_to_apply": version.when_to_apply or "",
            },
            # The text it changed TO, beside the text it changed from, so the
            # client can render the diff without a second call per row. A
            # listing that needs N follow-ups to say what it means is one
            # nobody scrolls.
            "current": {
                "title": rule.title,
                "statement": rule.statement or "",
                "when_to_apply": rule.when_to_apply or "",
            },
        }
        if rule.arose_from_id and rule.arose_from_id in titles:
            row["taught_by"] = {
                "id": rule.arose_from_id, "title": titles[rule.arose_from_id],
            }
        out.append(row)
    return out


async def restore_rule_version(
    rule_id: int, version_id: int, user_id: int,
) -> Optional[Rule]:
    """Put a preference back to what it said. None when it is not readable.

    MILESTONE 323 DECIDED THE OPPOSITE FOR RULES, and that decision stands —
    `routes/rulebooks.py` still has no restore for them. Its reasoning:
    "a binding instruction should not be revertible in one click", because a
    silent revert erases the only record of why the rewrite happened.

    A preference inverts both halves of that. The rewrite was not the
    operator's — the agent makes it mid-work, without asking, which is the
    whole design — so reverting is not undoing their own decision but
    exercising a veto over someone else's. And the veto has to be cheaper
    than shrugging, or drift is only nominally supervised.

    What makes it safe is that nothing is erased. This goes through
    `update_rule` like any other edit, so the restore takes its own snapshot:
    the rewrite stays in the history, with the revert recorded after it. The
    history gains an entry rather than losing one.

    Raises ValueError on a rule, so the kind check cannot be forgotten by a
    caller that reaches this directly.
    """
    async with async_session() as session:
        rule = await _fetch_owned_rule(session, rule_id, user_id)
        if rule is None:
            return None
        if (rule.kind or "rule") != "preference":
            raise ValueError(
                "only a preference can be restored in one action. A rule is "
                "the operator's own decision and reverting it goes through an "
                "ordinary edit, so the reason for the rewrite stays visible "
                "(milestone 323)."
            )
        version = (await session.execute(
            select(RuleVersion).where(
                RuleVersion.id == version_id, RuleVersion.rule_id == rule_id,
            )
        )).scalar_one_or_none()
        if version is None:
            return None
        fields = {
            "title": version.title,
            "statement": version.statement,
            "when_to_apply": version.when_to_apply,
            "why": version.why,
            "how_to_apply": version.how_to_apply,
        }

    # Outside the session above, because update_rule opens its own — and
    # through it rather than beside it, so the snapshot, the trigger guard and
    # the embedding refresh all happen exactly as they do for a hand edit.
    # A `None` field in the snapshot means the preference had nothing there,
    # so it is CLEARED rather than left at today's value; passing None to
    # update_rule means "leave alone", which would half-restore it.
    clear = [k for k, v in fields.items() if not (v or "").strip()]
    return await update_rule(
        rule_id, user_id,
        clear=[k for k in clear if k != "statement"],
        **{k: v for k, v in fields.items() if v},
    )


# ── 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 set_rule_moments(
    rule_id: int, user_id: int, moments: list[str],
) -> list[str] | None:
    """Replace which MOMENTS a rule arrives at (milestone 458). None if not owned.

    Set-semantics like set_rule_systems: the list given IS the state after.
    Every name is checked against the catalog first and the whole write is
    refused on the first unknown one — a typo stored as a mount would read
    back as attached and never fire, the silent miss moments exist to end.
    """
    from scribe.models.rulebook import rule_moments as rule_moments_t
    from scribe.services.moments import require_moments

    wanted = require_moments(moments) or []
    async with async_session() as session:
        rule = await _fetch_owned_rule(session, rule_id, user_id)
        if rule is None:
            return None
        await session.execute(
            sql_delete(rule_moments_t).where(rule_moments_t.c.rule_id == rule_id)
        )
        for moment in wanted:
            await session.execute(
                insert(rule_moments_t).values(rule_id=rule_id, moment=moment)
            )
        await session.commit()
        return wanted


async def list_rule_moments(rule_ids: list[int]) -> dict[int, list[str]]:
    """The moments each of a batch of rules is mounted on, in catalog order."""
    from scribe.models.rulebook import rule_moments as rule_moments_t
    from scribe.services.moments import MOMENTS

    if not rule_ids:
        return {}
    async with async_session() as session:
        rows = (await session.execute(
            select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
            .where(rule_moments_t.c.rule_id.in_(rule_ids))
        )).all()
    order = {name: i for i, name in enumerate(MOMENTS)}
    out: dict[int, list[str]] = {}
    for rule_id, moment in rows:
        out.setdefault(rule_id, []).append(moment)
    for names in out.values():
        names.sort(key=lambda n: (order.get(n, len(order)), n))
    return out


async def rules_on_moments(
    user_id: int, moments: list[str], project_id: int | None = None,
) -> list[tuple[Rule, str]]:
    """The rules mounted on any of these moments that this caller may be shown.

    The delivery read (milestone 458 step 4), and a LOOKUP: no score, no bar.
    Scoped by the same home clause the semantic search uses (rule_scope), so
    a mount never delivers another project's rule. Each rule appears once,
    under the first of `moments` it is mounted on — the caller passes them in
    catalog order, so that is the most specific moment the work reached.
    """
    from scribe.models.rulebook import rule_moments as rule_moments_t
    from scribe.services.rule_scope import joined_to_homes, rule_home

    if not moments:
        return []
    home = await rule_home(user_id, project_id)
    async with async_session() as session:
        rows = (await session.execute(
            joined_to_homes(
                select(Rule, rule_moments_t.c.moment)
                .select_from(rule_moments_t)
                .join(Rule, Rule.id == rule_moments_t.c.rule_id)
            )
            .where(
                rule_moments_t.c.moment.in_(list(moments)),
                Rule.deleted_at.is_(None),
                home,
            )
            .order_by(Rule.id)
        )).all()
    rank = {m: i for i, m in enumerate(moments)}
    first: dict[int, tuple[Rule, str]] = {}
    for rule, moment in rows:
        held = first.get(rule.id)
        if held is None or rank[moment] < rank[held[1]]:
            first[rule.id] = (rule, moment)
    return sorted(first.values(), key=lambda pair: (rank[pair[1]], pair[0].id))


async def mounted_moments(user_id: int) -> set[str]:
    """Every moment at least one of this caller's live rules is mounted on.

    In ANY home (rule_scope's `everywhere`): this answers "can a call reach
    anything at all", which is what lets the plugin stay off the wire for a
    tool whose moments carry nothing. A superset is the safe direction — the
    delivery read still scopes by project.
    """
    return set(await mount_counts(user_id))


async def mount_counts(user_id: int) -> dict[str, int]:
    """How many of this caller's live rules are mounted on each moment.

    The same read as `mounted_moments`, counted — what the Settings view
    shows beside each moment, so a moment carrying nothing reads as such.
    Moments with no mount are absent rather than zero.
    """
    from scribe.models.rulebook import rule_moments as rule_moments_t
    from scribe.services.rule_scope import joined_to_homes, rule_home

    home = await rule_home(user_id, everywhere=True)
    async with async_session() as session:
        rows = (await session.execute(
            joined_to_homes(
                select(rule_moments_t.c.moment, func.count(func.distinct(Rule.id)))
                .select_from(rule_moments_t)
                .join(Rule, Rule.id == rule_moments_t.c.rule_id)
            )
            .where(Rule.deleted_at.is_(None), home)
            .group_by(rule_moments_t.c.moment)
        )).all()
    return {moment: int(n) for moment, n in rows}


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()


# ── get_applicable_rules ────────────────────────────────────────────────

async def get_applicable_rules(
    project_id: int, user_id: int, limit: int = 50,
) -> dict:
    """The rules a project's LISTING shows: its own, and the global rules
    deterministically bound to the areas it works in.

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

    NOT WHAT A SESSION RECEIVES. Rules reach a session by retrieval, which
    reads a rule's home (milestone 414): global rules everywhere, a project's
    own rules in that project. This is the listing a planning read carries so
    a reader can see which constraints are on the table, and it is narrower
    than retrieval on purpose — "every global rule" is not a list anyone reads.

    `rules` is the global rules TAGGED to a canonical area this project works
    in (milestone 307, D7): a deterministic tag match, never a similarity
    score. Before milestone 414 this was every rule in a SUBSCRIBED rulebook,
    narrowed by area only where an author had tagged one. With subscriptions
    gone there is no opt-in left to scope the untagged ones, and an untagged
    global rule is general by construction — it arrives by retrieval when the
    work makes it relevant, like every other rule.

    `project_rules` is the project's own, never filtered by area: a rule
    written ON a project is scoped to it already.
    """
    from scribe.models.project import Project

    async with async_session() as session:
        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()

        rule_rows = []
        if project_area_ids:
            # Selects the ENTITY, not a column list: rule_brief is the one
            # place that decides which fields a surfaced rule carries. Filtered
            # in SQL so `limit` counts the rules that will actually be shown.
            rule_rows = (await session.execute(
                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)
                .where(
                    Rulebook.owner_user_id == user_id,
                    Rule.deleted_at.is_(None),
                    RulebookTopic.deleted_at.is_(None),
                    Rulebook.deleted_at.is_(None),
                    Rule.id.in_(
                        select(rule_systems.c.rule_id).where(
                            rule_systems.c.canonical_id.in_(project_area_ids)
                        )
                    ),
                )
                .order_by(
                    Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
                )
                .limit(limit + 1)
            )).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.
        proj_rule_rows = (await session.execute(
            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)
        )).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 — but only a partner that could reach this
    # project at all. An edge to another project's rule is not a way in.
    surfaced_ids = [r["id"] for r in rules] + [r["id"] for r in project_rules]
    partners = await co_surfaced_partners(user_id, surfaced_ids)
    for partner in partners:
        if partner.project_id not in (None, project_id):
            continue
        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 global rule is in this listing 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, "truncated": truncated}


def rules_payload(
    applicable: dict, *, user_id: int | None, source: str, brief: bool = False,
) -> 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 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.

    IT ALSO RECORDS THE SURFACING, which is why it 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
    other caller of the rules machinery — the write-path etag arm
    (`plugin_context`) — computes a marker and shows nobody anything, and
    counting it would put rules in the denominator that no agent ever saw.

    EVERY FORM LISTS, NONE RESTATES. Rules reach a session in full by
    retrieval, so these payloads say which constraints exist — id and title,
    the topic a global rule sits in, `via` for a co_surfaces partner — and
    get_rule reads one. Planning reads carried the full rule_brief until a
    project's listing grew to every global rule tagged to its areas
    (milestone 414) and start_planning replied with 92k characters (#4081),
    the shape #4045 had just removed from the handshake.

    `brief` is the session handshake's form (#4045): the project's own rules
    only. Only what is shown is recorded as surfaced.
    """
    project_rules = [_rule_line(r) for r in applicable.get("project_rules", [])]
    if brief:
        record_rule_surfaced(
            user_id=user_id, rule_ids=[r["id"] for r in project_rules], source=source,
        )
        return {"project_rules": project_rules}
    rules = [_rule_line(r) for r in applicable.get("rules", [])]
    record_rule_surfaced(
        user_id=user_id,
        rule_ids=[r["id"] for r in rules] + [r["id"] for r in project_rules],
        source=source,
    )
    return {
        "applicable_rules": rules,
        "applicable_rules_truncated": applicable["truncated"],
        "project_rules": project_rules,
    }


def _rule_line(brief: dict) -> dict:
    """One rule as a listing names it: enough to recognise it and fetch it.
    Keys a row does not carry are left out rather than sent empty (#2483)."""
    line = {"id": brief["id"], "title": brief["title"]}
    for key in ("topic_title", "via"):
        if brief.get(key):
            line[key] = brief[key]
    return line


# ── 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"


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

async def rules_due_for_verification(
    user_id: int,
    older_than_days: int = 0,
    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.
        never_only: only rules that have never been verified.
    """
    from datetime import datetime, timedelta, timezone

    from scribe.models.project import Project

    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 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,
        "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