"""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, 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"}
        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) -> 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 = "", 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,
            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).
    """
    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
        # 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
        # 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()


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


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

    `brief` is the session handshake's form (#4045): the project's own rules
    as id and title, nothing else. Rules reach a session in full by
    retrieval, so the handshake lists which of the project's constraints exist
    rather than restating them; get_rule reads one. Only what is shown is
    recorded as surfaced.
    """
    if brief:
        project_rules = [
            {"id": r["id"], "title": r["title"]}
            for r in applicable.get("project_rules", [])
        ]
        record_rule_surfaced(
            user_id=user_id, rule_ids=[r["id"] for r in project_rules], source=source,
        )
        return {"project_rules": project_rules}
    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"],
        "project_rules": applicable.get("project_rules", []),
    }


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