FabledGrimoire/src/scribe/services/lesson_rules.py

"""Lessons point at rules (milestone 440, #4196).

A lesson is a non-binding record of one situation; a rule is the binding
choice for a class of them. This service owns the link between the two — which
rule a lesson is an instance of — and both directions of reading it.

THE STATES (models/lesson_rule_link.py says why each exists):
`suggested` while evidence accumulates, `confirmed` or `rejected` once a
judgment is made. Every write here is a JUDGMENT, so every write lands as
confirmed or rejected; `suggested` rows come from the co-surfacing recorder
(#4637), never from a caller naming a rule.

THE THIRD ANSWER (#4631). "Which rule is this an instance of?" has three
answers, and only two of them are links: a rule named (confirmed), or NO RULE
FITS — a `lesson_no_rule` row carrying the reason. A lesson with neither is
UNJUDGED, and `list_unjudged` lists those. The three are kept apart because a
lesson that stands alone is the raw material for a rule nobody has written yet
(#4634), and that is unreadable if "nobody looked" means the same thing.

THE SOFT LINK (#4637). When one hook request puts a lesson and a rule in
front of the reader together, `co_surfaced` records the pair as evidence on a
SUGGESTED link, and once that evidence crosses a bar the next co-arrival asks
the reader to judge it. The evidence counts distinct SITUATIONS, not
arrivals: two records that resemble each other arrive together on every
similar prompt, and counting each time would measure their similarity over
and over rather than any relation between them.

ACL (rule 78). Linking changes what a lesson says about itself, so it needs
WRITE on the lesson (`access.can_write_note`, share-aware). It names a rule, so
it needs the rule to be one the caller may read — rules are owner-scoped, and
`rulebooks._fetch_owned_rule` / `_owned_rules_clause` are that check's two
forms. A read shows only the rules the READER owns: a lesson shared with
someone must not hand them the titles of its owner's private rules.
"""
from __future__ import annotations

import hashlib
import logging
import re
from datetime import datetime, timedelta, timezone

from sqlalchemy import and_, delete, exists, func, not_, or_, select
from sqlalchemy.exc import IntegrityError

from scribe.models import async_session
from scribe.models.lesson_rule_link import (
    CONFIRMED, REJECTED, SUGGESTED, LessonNoRule, LessonRuleLink,
)
from scribe.models.note import Note
from scribe.models.rulebook import Rule

logger = logging.getLogger(__name__)

# What a caller says to judge one pair. Words rather than the stored states,
# because "confirm" and "reject" are acts and the states are their results.
VERDICTS = {"confirm": CONFIRMED, "reject": REJECTED}

# Recorded on a link that set-semantics removed, so a reader of the rejected
# row can tell an explicit "not this rule" from a link dropped by a rewrite.
_REMOVED_NOTE = "removed from the lesson's rules by an update"

# Recorded on a confirmed link that a "no rule fits" answer overturned, ahead
# of that answer's reason.
_NO_RULE_NOTE = "no rule fits: "

# What `rule_judgment` reads on a lesson payload — the three answers.
LINKED, NO_RULE, UNJUDGED = "linked", "no_rule", "unjudged"

# THE SOFT-LINK EVIDENCE BAR (#4637) — defaults, stated as defaults (rules 32,
# 115): nothing about this install chose them.
#
# Three distinct situations before a pair is put to the reader. One is
# coincidence and two is a pattern only by courtesy; three different prompts
# or files that each brought the same lesson and rule up together is the
# smallest count that says the pairing is not a single resemblance repeated.
PROPOSE_SITUATIONS = 3
# A proposal the reader passed over is asked again only after this long, so
# one session's run of related prompts does not repeat the same question on
# every turn.
PROPOSE_COOLDOWN = timedelta(hours=6)
# Fingerprints kept per link. The bar needs three; the rest is for the reader
# judging later, and a list that grew with every situation would bloat a row
# whose answer stopped changing long ago.
_SITUATION_CAP = 50
# A token shorter than this is grammar, not situation — "a", "is", "to".
_MIN_TOKEN = 3

# How many rules a new lesson is offered to judge against. A handful, not the
# fifty `what_might_apply` returns: this is read in the create response, at
# the moment of writing, and a long tail there buries the one that fits.
CANDIDATE_LIMIT = 5


def _ids(values) -> list[int]:
    """Positive ints, de-duplicated, in the order given."""
    out: list[int] = []
    for v in values or []:
        try:
            i = int(v)
        except (TypeError, ValueError):
            raise ValueError(f"rule id {v!r} is not an integer")
        if i > 0 and i not in out:
            out.append(i)
    return out


async def _require_lesson_writable(user_id: int, lesson_id: int) -> None:
    from scribe.services import access
    from scribe.services import lessons as lessons_svc

    note = await lessons_svc.get_lesson(user_id, lesson_id)
    if note is None:
        raise ValueError(f"lesson {lesson_id} not found")
    if not await access.can_write_note(user_id, lesson_id):
        raise PermissionError(f"lesson {lesson_id} is not yours to change")


async def require_rules(user_id: int, rule_ids) -> list[int]:
    """The ids, validated as rules the caller owns — all of them or none.

    Checked BEFORE anything is written, by every caller, so a lesson create
    that names a rule it cannot see fails without leaving a half-linked
    lesson behind.
    """
    from scribe.services import rulebooks as rulebooks_svc

    wanted = _ids(rule_ids)
    missing = [rid for rid in wanted if await rulebooks_svc.get_rule(rid, user_id) is None]
    if missing:
        raise ValueError(
            f"rule(s) {missing} not found — a lesson can point only at a rule "
            "you can read. Nothing was linked."
        )
    return wanted


def require_one_answer(linked, no_rule: str) -> None:
    """Refuse "these rules fit" and "no rule fits" together, before any write
    — they contradict, and either order of applying them loses one."""
    if linked and (no_rule or "").strip():
        raise ValueError(
            "rule_ids and no_rule are the two answers to \"which rule is this "
            "an instance of?\" — give the one that holds. Nothing was written."
        )


async def _upsert(session, lesson_id: int, rule_id: int, state: str, note: str) -> None:
    now = datetime.now(timezone.utc)
    row = (await session.execute(
        select(LessonRuleLink).where(
            LessonRuleLink.lesson_id == lesson_id,
            LessonRuleLink.rule_id == rule_id,
        )
    )).scalar_one_or_none()
    if row is None:
        session.add(LessonRuleLink(
            lesson_id=lesson_id, rule_id=rule_id, state=state,
            note=note or None, judged_at=now,
        ))
        return
    # Evidence is kept across a judgment: a confirmation reads best beside
    # what it rested on.
    row.state = state
    row.note = note or row.note
    row.judged_at = now


async def _clear_no_rule(session, lesson_id: int) -> None:
    """A confirmed rule overturns "no rule fits" — both cannot be the answer."""
    await session.execute(delete(LessonNoRule).where(LessonNoRule.lesson_id == lesson_id))


async def set_lesson_rules(
    user_id: int, lesson_id: int, rule_ids, *, note: str = "",
) -> None:
    """Make the lesson's CONFIRMED rules exactly `rule_ids` (set-semantics).

    A rule named here is confirmed, whatever state it was in: the writer of a
    lesson saying "this is an instance of rule N" is the judgment the
    suggested state waits for, so it needs no evidence bar.

    A rule that WAS confirmed and is no longer named becomes `rejected`, not
    deleted. Dropping it is a judgment that the lesson is not an instance of
    that rule, and a deleted row would let the co-surfacing recorder propose
    the same pair again.
    """
    await _require_lesson_writable(user_id, lesson_id)
    wanted = await require_rules(user_id, rule_ids)
    async with async_session() as session:
        current = (await session.execute(
            select(LessonRuleLink).where(
                LessonRuleLink.lesson_id == lesson_id,
                LessonRuleLink.state == CONFIRMED,
            )
        )).scalars().all()
        for row in current:
            if row.rule_id not in wanted:
                row.state = REJECTED
                row.note = _REMOVED_NOTE
                row.judged_at = datetime.now(timezone.utc)
        for rid in wanted:
            await _upsert(session, lesson_id, rid, CONFIRMED, note)
        if wanted:
            await _clear_no_rule(session, lesson_id)
        await session.commit()


async def set_no_rule(user_id: int, lesson_id: int, why: str) -> dict:
    """Record that no rule governs this lesson's situation, and why.

    A confirmed link the lesson still carries becomes `rejected`, with the
    reason: saying none fits is saying the linked one does not either. A
    second answer replaces the first, so the reason on file is the latest.
    """
    why = (why or "").strip()
    if not why:
        raise ValueError(
            "no_rule takes the reason no rule fits, in a line — it is what "
            "lets the answer be re-judged when a rule is later written for "
            "this situation."
        )
    await _require_lesson_writable(user_id, lesson_id)
    now = datetime.now(timezone.utc)
    async with async_session() as session:
        confirmed = (await session.execute(
            select(LessonRuleLink).where(
                LessonRuleLink.lesson_id == lesson_id,
                LessonRuleLink.state == CONFIRMED,
            )
        )).scalars().all()
        for row in confirmed:
            row.state = REJECTED
            row.note = _NO_RULE_NOTE + why
            row.judged_at = now
        answer = await session.get(LessonNoRule, lesson_id)
        if answer is None:
            session.add(LessonNoRule(lesson_id=lesson_id, why=why, judged_at=now))
        else:
            answer.why = why
            answer.judged_at = now
        await session.commit()
    return {"why": why, "judged_at": now.isoformat()}


async def judge_link(
    user_id: int, lesson_id: int, rule_id: int, verdict: str, note: str = "",
) -> dict:
    """Confirm or reject one (lesson, rule) pair, suggested or not."""
    state = VERDICTS.get((verdict or "").strip().lower())
    if state is None:
        raise ValueError(f"verdict must be one of {sorted(VERDICTS)}, got {verdict!r}")
    await _require_lesson_writable(user_id, lesson_id)
    [rid] = await require_rules(user_id, [rule_id])
    async with async_session() as session:
        await _upsert(session, lesson_id, rid, state, note)
        if state == CONFIRMED:
            await _clear_no_rule(session, lesson_id)
        await session.commit()
        row = (await session.execute(
            select(LessonRuleLink).where(
                LessonRuleLink.lesson_id == lesson_id,
                LessonRuleLink.rule_id == rid,
            )
        )).scalar_one()
        return row.to_dict()


async def rules_for_lessons(user_id: int, lesson_ids) -> dict[int, list[dict]]:
    """{lesson_id: [{id, title, kind, state, note}]}, for rules the READER owns.

    One query for the whole set — a lesson list would otherwise be N+1.
    Confirmed first, then suggested, then rejected: the order a reader cares
    about them in.
    """
    from scribe.services.rulebooks import _owned_rules_clause

    ids = [int(i) for i in lesson_ids or []]
    out: dict[int, list[dict]] = {i: [] for i in ids}
    if not ids:
        return out
    order = {s: n for n, s in enumerate((CONFIRMED, SUGGESTED, REJECTED))}
    async with async_session() as session:
        rows = (await session.execute(
            select(LessonRuleLink, Rule)
            .join(Rule, Rule.id == LessonRuleLink.rule_id)
            .where(LessonRuleLink.lesson_id.in_(ids))
            .where(_owned_rules_clause(user_id))
        )).all()
    for link, rule in sorted(rows, key=lambda r: (order.get(r[0].state, 9), r[1].id)):
        item = {
            "id": rule.id, "title": rule.title, "kind": rule.kind,
            "state": link.state, "note": link.note or "",
        }
        if link.state == SUGGESTED:
            # What the suggestion rests on, so a reader can judge it from
            # here rather than having to go and count.
            item["evidence"] = evidence_summary(link.evidence)
        out[link.lesson_id].append(item)
    return out


async def lessons_for_rule(user_id: int, rule_id: int) -> list[dict]:
    """The lessons that point at one rule, readable by the caller (share-aware).

    The reverse direction, and the one a rule's page needs: the concrete
    situations that have been judged instances of it.
    """
    from scribe.services.access import readable_notes_clause

    async with async_session() as session:
        rows = (await session.execute(
            select(LessonRuleLink, Note)
            .join(Note, Note.id == LessonRuleLink.lesson_id)
            .where(LessonRuleLink.rule_id == int(rule_id))
            .where(Note.deleted_at.is_(None))
            .where(readable_notes_clause(user_id))
        )).all()
    order = {s: n for n, s in enumerate((CONFIRMED, SUGGESTED, REJECTED))}
    return [
        {"id": note.id, "title": note.title, "state": link.state, "note": link.note or ""}
        for link, note in sorted(rows, key=lambda r: (order.get(r[0].state, 9), r[1].id))
    ]


async def judgments_for_lessons(lesson_ids) -> dict[int, dict]:
    """{lesson_id: {"rule_judgment": linked|no_rule|unjudged, "no_rule"?}}.

    Read from the link and answer tables directly, NOT from the reader's view
    of the rules: a lesson shared with someone who cannot see its owner's rule
    is still a judged lesson, and calling it unjudged would invite them to
    judge it again.
    """
    ids = [int(i) for i in lesson_ids or []]
    if not ids:
        return {}
    async with async_session() as session:
        linked = set((await session.execute(
            select(LessonRuleLink.lesson_id).where(
                LessonRuleLink.lesson_id.in_(ids),
                LessonRuleLink.state == CONFIRMED,
            )
        )).scalars().all())
        answers = {
            a.lesson_id: a for a in (await session.execute(
                select(LessonNoRule).where(LessonNoRule.lesson_id.in_(ids))
            )).scalars().all()
        }
    out: dict[int, dict] = {}
    for i in ids:
        if i in linked:
            out[i] = {"rule_judgment": LINKED}
        elif i in answers:
            out[i] = {"rule_judgment": NO_RULE, "no_rule": answers[i].to_dict()}
        else:
            out[i] = {"rule_judgment": UNJUDGED}
    return out


async def attach_lesson_rules(user_id: int, rows: list[dict], *, key: str = "id") -> None:
    """Add `rules` and `rule_judgment` (plus `no_rule` when that is the answer)
    to each lesson payload row, in place — a fixed number of queries per page.

    Fail-open (snippet #4286): this decorates a lesson the caller already has,
    so a failed lookup leaves the keys off and logs, rather than refusing the
    lesson. An absent key reads as "not attached", never as "no rule".
    """
    ids = [int(r[key]) for r in rows if isinstance(r.get(key), int)]
    try:
        found = await rules_for_lessons(user_id, ids)
        judged = await judgments_for_lessons(ids)
    except Exception:
        logger.warning("lesson→rule links could not be read", exc_info=True)
        return
    for r in rows:
        if isinstance(r.get(key), int):
            r["rules"] = found.get(r[key], [])
            r.update(judged.get(r[key], {}))


def unjudged_clause():
    """SQL: a lesson (a `notes` row) with no confirmed rule and no "no rule
    fits" answer. A rejected link alone leaves it unjudged — "not that rule"
    does not say whether another one fits."""
    from scribe.models.note import Note

    return and_(
        not_(exists().where(and_(
            LessonRuleLink.lesson_id == Note.id,
            LessonRuleLink.state == CONFIRMED,
        ))),
        not_(exists().where(LessonNoRule.lesson_id == Note.id)),
    )


async def list_unjudged(
    user_id: int, *, tag: str = "", project_id: int | None = None,
    limit: int = 50, offset: int = 0,
) -> tuple[list[dict], int]:
    """The lessons nobody has answered "which rule?" for, newest edit first.

    The browse scope (`browsable_notes_clause`), the list it sits beside uses:
    a lesson shared directly with the caller is search-only and stays out of
    an ambient list. Rows come back in the knowledge list's item shape.
    """
    from scribe.models.note import Note
    from scribe.services.access import browsable_notes_clause
    from scribe.services.knowledge import _note_to_item
    from scribe.services.lessons import LESSON_NOTE_TYPE

    base = (
        select(Note)
        .where(browsable_notes_clause(user_id))
        .where(Note.note_type == LESSON_NOTE_TYPE)
        .where(Note.deleted_at.is_(None))
        .where(unjudged_clause())
    )
    if tag:
        base = base.where(Note.tags.contains([tag]))
    if project_id is not None:
        base = base.where(Note.project_id == project_id)
    async with async_session() as session:
        total = (await session.execute(
            select(func.count()).select_from(base.subquery())
        )).scalar_one()
        rows = (await session.execute(
            base.order_by(Note.updated_at.desc()).limit(limit).offset(offset)
        )).scalars().all()
    return [_note_to_item(n) for n in rows], int(total)


async def rule_candidates(
    user_id: int, what: str, when_to_apply: str, project_id: int | None = None,
) -> list[dict] | None:
    """The rules a lesson most resembles, for the writer to judge against.

    Searched with the lesson's claim and its trigger — the same two halves a
    rule's own document leads with (`trigger_title`), so like is
    compared with like; the insight is the story and would dilute the vector.
    Scoped as any project read is: global rules plus the lesson's project's
    own. The bar is the explicit rule search's, since this is an explicit
    question rather than an injection.

    None when the search could not run (no embedder, a failed query), so a
    caller can say "candidates unavailable" rather than "nothing resembles".
    """
    from scribe.services.embeddings import (
        DEFAULT_SIMILARITY_THRESHOLD, semantic_search_rules, trigger_title,
    )

    # The rule document's own title join, so like is compared with like.
    query = trigger_title(what, when_to_apply)
    report: dict = {}
    found = await semantic_search_rules(
        user_id, query, limit=CANDIDATE_LIMIT,
        threshold=DEFAULT_SIMILARITY_THRESHOLD, report=report,
        project_id=project_id or None,
    )
    if not report.get("searched"):
        return None
    return [
        {
            "id": rule.id, "title": rule.title, "kind": rule.kind,
            "when_to_apply": rule.when_to_apply or "", "score": round(score, 3),
        }
        for score, rule in found
    ]


async def attach_rule_lessons(user_id: int, data: dict, rule_id: int) -> None:
    """Add `lessons` to a rule payload, in place, when there are any.

    Present-only, as a rule's `systems` and `relations` are (#2483), and
    fail-open for the reason `attach_lesson_rules` gives.
    """
    try:
        lessons = await lessons_for_rule(user_id, rule_id)
    except Exception:
        logger.warning("rule→lesson links could not be read", exc_info=True)
        return
    if lessons:
        data["lessons"] = lessons



def situation_key(arm: str, text: str) -> str:
    """A fingerprint of one situation, so a repeat counts once.

    `arm` is part of the key — "p" for a prompt, "w" for a file being written
    — because the two name different kinds of situation and must not collide.
    On the prompt arm the text is the prompt: lowercased, split into word
    tokens of _MIN_TOKEN or more, de-duplicated and sorted, so the same ask
    re-sent with different spacing, punctuation, case or word order is one
    situation. On the write arm the text is the PATH, not the code: every
    edit to one file is one situation, however much the code differs between
    them. Empty when there is nothing to key on.
    """
    if arm == "w":
        basis = (text or "").strip()
    else:
        tokens = sorted({t for t in re.findall(r"[a-z0-9]+", (text or "").lower())
                         if len(t) >= _MIN_TOKEN})
        basis = " ".join(tokens)
    if not basis:
        return ""
    return f"{arm}:" + hashlib.sha1(basis.encode()).hexdigest()[:16]


def evidence_summary(evidence) -> dict:
    """The counts a reader judges a suggested link by."""
    ev = evidence if isinstance(evidence, dict) else {}
    return {
        "situations": len(ev.get("situations") or []),
        "projects": len(ev.get("projects") or []),
        "co_surfaced": int(ev.get("co_surfaced") or 0),
    }


def add_evidence(evidence, key: str, project_id: int | None, now: datetime) -> dict:
    """A NEW evidence dict with this co-arrival counted. Pure, so the counting
    rules are testable without a database; a new dict, so SQLAlchemy sees the
    JSONB column change."""
    ev = dict(evidence) if isinstance(evidence, dict) else {}
    ev["co_surfaced"] = int(ev.get("co_surfaced") or 0) + 1
    situations = list(ev.get("situations") or [])
    if key and key not in situations and len(situations) < _SITUATION_CAP:
        situations.append(key)
    ev["situations"] = situations
    projects = list(ev.get("projects") or [])
    if project_id and int(project_id) not in projects:
        projects.append(int(project_id))
    ev["projects"] = projects
    ev.setdefault("first_at", now.isoformat())
    ev["last_at"] = now.isoformat()
    return ev


def proposal_due(evidence, now: datetime) -> bool:
    """At the bar, and not asked within the cooldown."""
    ev = evidence if isinstance(evidence, dict) else {}
    if len(ev.get("situations") or []) < PROPOSE_SITUATIONS:
        return False
    last = ev.get("proposed_at")
    if not last:
        return True
    try:
        return now - datetime.fromisoformat(last) >= PROPOSE_COOLDOWN
    except (TypeError, ValueError):
        return True


def _proposal_line(lesson_id: int, lesson_title: str, rule_id: int, rule_title: str,
                   kind: str, evidence: dict) -> str:
    counts = evidence_summary(evidence)
    where = (f" across {counts['projects']} projects" if counts["projects"] > 1 else "")
    return (
        f"> Lesson #{lesson_id} \"{lesson_title}\" and {kind} #{rule_id} "
        f"\"{rule_title}\" keep arriving together — {counts['situations']} "
        f"distinct situations{where}. If the lesson is an instance of the "
        f"{kind}, `judge_lesson_link({lesson_id}, {rule_id}, \"confirm\", "
        f"note=\"why\")` links them and the {kind} becomes reachable through "
        f"the lesson's situation; if it is not, `\"reject\"` with the why "
        f"stops the pair being asked about again."
    )


async def co_surfaced(
    user_id: int, lesson_ids, rule_ids, *, arm: str, situation: str,
    project_id: int | None = None,
) -> str:
    """Record that these lessons and rules arrived together in one request,
    and return a proposal line when a pair has earned one ("" otherwise).

    Called by the hook doors that put both kinds in front of the reader in
    the SAME response — `/retrieve` on a prompt, `/prior-art` on a write — so
    the pairing is exact: nothing is joined across requests, and no session
    identity is needed.

    Only pairs the reader could confirm are recorded: a lesson they may write
    and a rule they own, the two checks `judge_link` makes. A pair already
    judged (confirmed or rejected) gathers nothing more — a confirmation needs
    no evidence, and a rejection is what stops the asking.

    One proposal per request at most, so a menu never becomes a queue of
    questions. Fails open: this rides a hook, and a recall aid must never
    break the prompt or the write it decorates.
    """
    try:
        return await _co_surfaced(user_id, lesson_ids, rule_ids, arm, situation, project_id)
    except Exception:
        logger.warning("lesson/rule co-surfacing could not be recorded", exc_info=True)
        return ""


async def _co_surfaced(user_id, lesson_ids, rule_ids, arm, situation, project_id) -> str:
    from scribe.services.access import can_write_note
    from scribe.services.rulebooks import _owned_rules_clause

    lessons = _ids(lesson_ids)
    rules = _ids(rule_ids)
    key = situation_key(arm, situation)
    if not lessons or not rules or not key:
        return ""
    lessons = [lid for lid in lessons if await can_write_note(user_id, lid)]
    if not lessons:
        return ""
    now = datetime.now(timezone.utc)
    proposal = None
    async with async_session() as session:
        owned = {
            r.id: r for r in (await session.execute(
                select(Rule).where(Rule.id.in_(rules)).where(_owned_rules_clause(user_id))
            )).scalars().all()
        }
        if not owned:
            return ""
        existing = {
            (row.lesson_id, row.rule_id): row for row in (await session.execute(
                select(LessonRuleLink).where(
                    LessonRuleLink.lesson_id.in_(lessons),
                    LessonRuleLink.rule_id.in_(list(owned)),
                )
            )).scalars().all()
        }
        for lid in lessons:
            for rid in owned:
                row = existing.get((lid, rid))
                if row is not None and row.state != SUGGESTED:
                    continue
                if row is None:
                    row = LessonRuleLink(lesson_id=lid, rule_id=rid, state=SUGGESTED)
                    session.add(row)
                ev = add_evidence(row.evidence, key, project_id, now)
                if proposal is None and proposal_due(ev, now):
                    ev["proposed_at"] = now.isoformat()
                    ev["proposed_count"] = int(ev.get("proposed_count") or 0) + 1
                    proposal = (lid, rid, ev)
                row.evidence = ev
        try:
            await session.commit()
        except IntegrityError:
            # Two hook requests created the same pair at once; the other
            # one's row stands, and this co-arrival is simply not counted.
            await session.rollback()
            return ""
        if proposal is None:
            return ""
        lid, rid, ev = proposal
        lesson_title = (await session.execute(
            select(Note.title).where(Note.id == lid)
        )).scalar_one_or_none() or ""
    rule = owned[rid]
    return _proposal_line(lid, lesson_title, rid, rule.title, rule.kind or "rule", ev)


# ── Reaching a rule through its lessons (#4633) ──────────────────────────────


async def confirmed_lessons(user_id: int) -> set[int]:
    """The lessons, readable by the caller, that carry a CONFIRMED link to a
    rule the caller owns — the only lessons that can bring a rule along.

    CONFIRMED ONLY, and the reason is the soft link's: a suggested link that
    carried its rule would put the two side by side on every prompt that
    found the lesson, which is exactly the co-arrival #4637 counts as
    evidence — the suggestion would manufacture its own proof.

    Asked first and cheaply, so an arm on an install with no confirmed links
    — every install, until someone judges one — never pays for a search.
    """
    from scribe.services.access import readable_notes_clause
    from scribe.services.rulebooks import _owned_rules_clause

    async with async_session() as session:
        rows = (await session.execute(
            select(LessonRuleLink.lesson_id)
            .join(Rule, Rule.id == LessonRuleLink.rule_id)
            .join(Note, Note.id == LessonRuleLink.lesson_id)
            .where(LessonRuleLink.state == CONFIRMED)
            .where(Rule.deleted_at.is_(None))
            .where(_owned_rules_clause(user_id))
            .where(Note.deleted_at.is_(None))
            .where(readable_notes_clause(user_id))
            .distinct()
        )).scalars().all()
    return {int(i) for i in rows}


async def confirmed_rules_in_scope(
    user_id: int, lesson_ids, project_id: int | None,
) -> dict[int, list]:
    """{lesson_id: [Rule]} — each lesson's CONFIRMED rules that may surface here.

    "May surface here" is the rule arms' own scope (milestone 414): a global
    rule, or a rule of THIS project. A lesson is project-independent and
    surfaces everywhere, so without this a lesson linked to project A's rule
    would carry that rule into project B's sessions — the leak milestone 414
    closed for the direct arms, reopened through a side door.
    """
    from scribe.services.rulebooks import _owned_rules_clause

    ids = [int(i) for i in lesson_ids or []]
    if not ids:
        return {}
    home = (
        or_(Rule.topic_id.isnot(None), Rule.project_id == project_id)
        if project_id else Rule.topic_id.isnot(None)
    )
    async with async_session() as session:
        rows = (await session.execute(
            select(LessonRuleLink.lesson_id, Rule)
            .join(Rule, Rule.id == LessonRuleLink.rule_id)
            .where(LessonRuleLink.lesson_id.in_(ids))
            .where(LessonRuleLink.state == CONFIRMED)
            .where(Rule.deleted_at.is_(None))
            .where(_owned_rules_clause(user_id))
            .where(home)
            .order_by(Rule.id)
        )).all()
    out: dict[int, list] = {}
    for lesson_id, rule in rows:
        out.setdefault(int(lesson_id), []).append(rule)
    return out