FabledScribe/src/scribe/services/notes.py

import logging
import re
from collections.abc import Callable, Iterable
from datetime import date, datetime, timezone

from sqlalchemy import func, or_, select, text

from scribe.models import async_session
from scribe.models.note import Note, TaskKind, TaskPriority, TaskStatus
from scribe.models.base import iso

logger = logging.getLogger(__name__)

# The fields a derived `data` mirror is parsed out of. Writing any of them can
# change what the mirror should say, so update_note recomposes it when one
# moves. Kept here as a set of NAMES rather than imported, because it
# describes update_note's own `fields` dict, not any parser's signature.
_PARSED_FROM_BODY = frozenset({"title", "body", "tags"})


def _mirror_recomposers() -> dict[str, Callable[[Note], dict]]:
    """note_type -> the function that rebuilds that kind's derived `data`.

    A TABLE rather than a chain of `if note_type == ...`, because the previous
    shape tested one constant and the next kind with a derived mirror was
    silently not covered — which is exactly what happened: the snippet guard
    (#3128) was hard-coded, and lessons arrived in milestone 385 with the same
    body-is-authority/`data`-is-mirror design and none of the protection.

    The failure is invisible from here. Nothing raises, nothing logs; the row
    simply keeps answering queries from a mirror that no longer matches its
    body, and every surface that prefers the mirror — which is all of them, by
    design, because parsing markdown to answer what an index can answer is how
    a hot path rots — reports the old value confidently.

    Imported inside the function, not at module scope: both services call back
    into this module (`update_snippet`/`update_lesson` -> `update_note`), so a
    top-level import is a cycle.
    """
    from scribe.services.lessons import (
        LESSON_NOTE_TYPE, recompose_data as _lesson_mirror,
    )
    from scribe.services.snippets import (
        SNIPPET_NOTE_TYPE, recompose_data as _snippet_mirror,
    )
    return {
        SNIPPET_NOTE_TYPE: _snippet_mirror,
        LESSON_NOTE_TYPE: _lesson_mirror,
    }


# Text fields where EMPTY MEANS NULL (milestone 317). The sweep's whole signal
# is `verify_with IS NULL` = "this is a decision, there is nothing to go and
# check". An empty string that is not NULL makes a norm look like a constraint
# nobody has verified, forever — and it would sit at the top of the sweep,
# since never-checked sorts first. Sibling of rulebooks.NULLABLE_RULE_TEXT.
NULLABLE_NOTE_TEXT = ("verify_with", "expires_when")


def guard_check_fields(status: str | None, note_type: str | None) -> None:
    """Raise unless a record in this shape may carry verify_with/expires_when.

    Stated as an INVARIANT over the resulting record rather than a filter on
    which fields a caller passed, so it also catches the sideways route: a
    checked note being turned into a task, which no per-field gate would see.

    Raises rather than dropping silently, for minted_kind's reason (#3129) — a
    silently-corrected write is the defect that reasoning exists to end, and a
    caller putting a check on the wrong record has an idea an error corrects
    and a default hides. Lives at the service, not either door, so REST and MCP
    cannot come to disagree about it.
    """
    if status is not None:
        raise ValueError(
            "a task cannot carry verify_with/expires_when: a task's decay is "
            "its status, and a done issue records what happened rather than "
            "asserting something that can later go false. Put the check on the "
            "note the fact lives in, or clear the check before making this a "
            "task."
        )
    if note_type == "snippet":
        raise ValueError(
            "a snippet already has a check: verify_snippet(), which compares "
            "the recorded location and code against the repo and expires its "
            "own verdict when the code moves. verify_with/expires_when are the "
            "free-text form, for prose notes that assert a fact about "
            "something outside the repo."
        )


def embed_note(note) -> None:
    """Refresh a note's embedding, fire-and-forget.

    Lives HERE — at the service, not the route — so every caller gets it by
    construction. Previously each REST route made this call itself and the MCP
    tools did not, so a record created through MCP stayed out of semantic search
    and auto-inject until the next restart's backfill ran (#2056). That is
    invisible on an instance that redeploys constantly and permanent on one that
    doesn't, which is the worst shape a bug can have: it only appears where
    nobody is looking.

    Uses `note.user_id` — the OWNER — rather than the caller. Embeddings belong
    to the record, and a collaborator editing a shared note must refresh the
    owner's row rather than mint a second one under their own id.

    Import is lazy so importing this module doesn't pull in the embedding model;
    exceptions are swallowed because a record that saved must not fail on its
    index refresh. No running loop (unit tests, scripts) is an ordinary case,
    not an error.

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

        from scribe.services.embeddings import document_title, upsert_note_embedding
        # Chunking and the empty-record gate live inside upsert_note_embedding —
        # one path for every writer (#280). The title is the EMBEDDED one: a
        # snippet's or lesson's trigger joins its name here, not in the stored
        # title (milestone 427).
        asyncio.create_task(
            upsert_note_embedding(
                note.id, note.user_id,
                document_title(note.title, note.note_type, note.data, note.body),
                note.body,
            )
        )
    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 note %s", note.id)


def _normalize_tags(tags: list[str]) -> list[str]:
    """Lowercase, strip, deduplicate, and drop empty tags."""
    seen: set[str] = set()
    out: list[str] = []
    for t in tags:
        normalized = t.strip().lower()
        if normalized and normalized not in seen:
            seen.add(normalized)
            out.append(normalized)
    return out


# Type-nouns the LLM tends to include in search queries. Treating them as
# required ILIKE terms drops literal-title matches; we strip them server-side
# and let the `type` / `project` parameters scope results instead.
_SEARCH_TYPE_NOUNS = {"task", "tasks", "note", "notes", "project", "projects"}


def _strip_type_nouns(q: str) -> list[str]:
    """Return q's tokens with type-nouns removed (case-insensitive)."""
    return [t for t in q.split() if t.lower() not in _SEARCH_TYPE_NOUNS]


async def _maybe_reactivate_project(project_id: int) -> None:
    """If a project is paused, reactivate it — activity indicates resumed work."""
    from scribe.models.project import Project
    try:
        async with async_session() as session:
            project = (await session.execute(
                select(Project).where(Project.id == project_id)
            )).scalars().first()
            if project and project.status == "paused":
                project.status = "active"
                await session.commit()
                logger.info("Auto-reactivated paused project %d (%s)", project_id, project.title)
    except Exception:
        logger.debug("_maybe_reactivate_project failed for project %d", project_id, exc_info=True)


async def create_note(
    user_id: int,
    title: str = "",
    body: str = "",
    description: str | None = None,
    tags: list[str] | None = None,
    parent_id: int | None = None,
    project_id: int | None = None,
    milestone_id: int | None = None,
    status: str | None = None,
    priority: str | None = None,
    due_date: date | None = None,
    recurrence_rule: dict | None = None,
    note_type: str = "note",
    task_kind: str = "work",
    arose_from_id: int | None = None,
    data: dict | None = None,
    verify_with: str | None = None,
    expires_when: str | None = None,
) -> Note:
    # Auto-populate project_id from milestone when not explicitly provided
    if milestone_id is not None and project_id is None:
        from scribe.models.milestone import Milestone
        async with async_session() as lookup:
            result = await lookup.execute(
                select(Milestone).where(Milestone.id == milestone_id, Milestone.user_id == user_id)
            )
            ms = result.scalars().first()
            if ms is not None:
                project_id = ms.project_id

    note = build_note(
        user_id,
        title=title,
        body=body,
        description=description,
        tags=tags,
        parent_id=parent_id,
        project_id=project_id,
        milestone_id=milestone_id,
        status=status,
        priority=priority,
        due_date=due_date,
        recurrence_rule=recurrence_rule,
        note_type=note_type,
        task_kind=task_kind,
        arose_from_id=arose_from_id,
        data=data,
        verify_with=verify_with,
        expires_when=expires_when,
    )
    async with async_session() as session:
        session.add(note)
        await session.commit()
        await session.refresh(note)

    embed_note(note)

    if project_id is not None:
        await _maybe_reactivate_project(project_id)

    return note


def build_note(
    user_id: int,
    title: str = "",
    body: str = "",
    description: str | None = None,
    tags: list[str] | None = None,
    parent_id: int | None = None,
    project_id: int | None = None,
    milestone_id: int | None = None,
    status: str | None = None,
    priority: str | None = None,
    due_date: date | None = None,
    recurrence_rule: dict | None = None,
    note_type: str = "note",
    task_kind: str = "work",
    arose_from_id: int | None = None,
    data: dict | None = None,
    verify_with: str | None = None,
    expires_when: str | None = None,
) -> Note:
    """A validated, UNSAVED Note — every check create_note makes, no write.

    Split out of create_note so a caller that must add several records in one
    transaction (services/record_batch.py) validates them exactly as a single
    create would, instead of carrying a second copy of the rules that drifts.
    Raises before anything is built, so nothing illegal reaches a session.
    """
    # Empty means empty (NULLABLE_NOTE_TEXT), then the invariant. Both run
    # before anything is written, so an illegal shape never reaches the table.
    verify_with = verify_with or None
    expires_when = expires_when or None
    if verify_with or expires_when:
        guard_check_fields(status, note_type)

    # Validate status/priority here so the MCP create_task path (which passes
    # them straight through) can't persist an out-of-enum value that the REST
    # route would have rejected — there's no DB CHECK on notes.status.
    if isinstance(status, str):
        try:
            status = TaskStatus(status).value
        except ValueError:
            raise ValueError(f"Invalid status: {status!r}. Must be one of: {[s.value for s in TaskStatus]}")
    if isinstance(priority, str):
        try:
            priority = TaskPriority(priority).value
        except ValueError:
            raise ValueError(f"Invalid priority: {priority!r}. Must be one of: {[p.value for p in TaskPriority]}")

    note = Note(
        user_id=user_id,
        title=title,
        body=body,
        description=description,
        tags=_normalize_tags(tags or []),
        parent_id=parent_id,
        project_id=project_id,
        milestone_id=milestone_id,
        status=status,
        priority=priority,
        due_date=due_date,
        recurrence_rule=recurrence_rule,
        note_type=note_type,
        task_kind=task_kind,
        arose_from_id=arose_from_id,
        data=data,
        verify_with=verify_with,
        expires_when=expires_when,
    )
    # A create that names a status is the same transition an update to it is
    # (#3683) — otherwise `create_task(status="in_progress")` writes a row no
    # update could produce: started, with no `started_at`.
    if status is not None:
        apply_status_transition(note, user_id)
    return note


def apply_status_transition(note: Note, user_id: int | None = None) -> None:
    """Stamp what reaching `note.status` implies — the ONE statement of it.

    Called by the update path whenever `status` is written and by `build_note`
    whenever a create names one, so a task created at a status is
    indistinguishable from one that reached it by update. Two copies of "what
    a status implies" are where the two drift, and #3683 was that drift.

    `user_id` is whoever is making the change: reaching `in_progress` is the
    moment a session takes the work on, so it stamps that user's claim, and a
    status that ends or un-starts the work releases it (milestone 381).
    """
    from scribe.services.task_claims import release_claim, stamp_claim

    _now = datetime.now(timezone.utc)
    if note.status == TaskStatus.in_progress.value:
        if note.started_at is None:
            note.started_at = _now
        if user_id is not None:
            stamp_claim(note, user_id, _now)
    elif note.status in (TaskStatus.done.value, TaskStatus.cancelled.value):
        release_claim(note)
        note.completed_at = _now
        if note.recurrence_rule:
            from scribe.services.recurrence import calculate_next_due
            base = note.due_date or _now.date()
            next_due = calculate_next_due(note.recurrence_rule, base)
            note.recurrence_next_spawn_at = datetime(
                next_due.year, next_due.month, next_due.day, tzinfo=timezone.utc
            )
    elif note.status == TaskStatus.todo.value:
        release_claim(note)
        note.started_at = None
        note.completed_at = None
        note.recurrence_next_spawn_at = None


async def get_note(user_id: int, note_id: int) -> Note | None:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(
                Note.id == note_id, Note.user_id == user_id, Note.deleted_at.is_(None)
            )
        )
        return result.scalars().first()


async def list_notes(
    user_id: int,
    q: str | None = None,
    tags: list[str] | None = None,
    is_task: bool | None = None,
    status: str | list[str] | None = None,
    priority: str | list[str] | None = None,
    due_before: date | None = None,
    due_after: date | None = None,
    project_id: int | None = None,
    milestone_id: int | None = None,
    milestone_ids: list[int] | None = None,
    parent_id: int | None = None,
    task_kind: str | None = None,
    no_project: bool = False,
    exclude_paused_projects: bool = False,
    sort: str = "updated_at",
    order: str = "desc",
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[Note], int]:
    """Lifecycle-shaped listing. Two contracts worth knowing:

    VISIBILITY is the shared browse clause (#47, #2462): the caller's own
    records plus anything in a project they can reach — the same reach
    query_knowledge has always had. Before this, list_notes was silently
    owner-only, so a task in a shared project was invisible in list_tasks,
    enter_project's open-task list and the web UI while the same project's
    notes appeared. Browse, not read, per decision #2094: an ambient list must
    never surface a record someone shared one-to-one — those stay search-only.

    `q` is SEMANTIC (operator decision, 2026-08-06: "make it match") — the same
    meaning-based match as Browse search, at the interactive floor. When `q` is
    present, relevance ordering wins and `sort` is ignored; a query is a
    relevance claim and sorting its results by date would shuffle the answer.
    Falls back to ILIKE substring match only when the embedder is unavailable —
    degraded but never empty. Superseded records are not demoted here: the
    penalty reorders a top-k, and reordering a paginated, counted list would
    make page boundaries lie. The search surfaces carry the demotion.

    `sort` is a Note column name, or "touched": the later of updated_at and
    the record's newest work-log.
    """
    from scribe.services.access import notes_visibility_clause

    visible = notes_visibility_clause(user_id, "browse")
    async with async_session() as session:
        query = select(Note).where(visible, Note.deleted_at.is_(None))
        count_query = select(func.count(Note.id)).where(
            visible, Note.deleted_at.is_(None)
        )

        # Filter by task vs note
        if is_task is True:
            query = query.where(Note.status.isnot(None))
            count_query = count_query.where(Note.status.isnot(None))
        elif is_task is False:
            query = query.where(Note.status.is_(None))
            count_query = count_query.where(Note.status.is_(None))

        semantic_order = None
        if q:
            query_vec = None
            try:
                from scribe.services.embeddings import get_embedding
                query_vec = await get_embedding(q)
            except Exception:
                query_vec = None   # embedder down → keyword fallback below
            if query_vec is not None:
                from scribe.models.embedding import NoteEmbedding
                from scribe.services.embeddings import INTERACTIVE_SEARCH_THRESHOLD
                # Best-chunk-per-note as a correlated MIN, not a join (#280):
                # a note stores one embedding row PER CHUNK, so the plain join
                # this used to be would repeat a long note once per matching
                # chunk — duplicated list rows and a total that counts chunks.
                # This query is filter-heavy and paginated, never HNSW-bound,
                # so the scalar subquery costs what the join did.
                best_distance = (
                    select(
                        func.min(
                            NoteEmbedding.embedding.cosine_distance(query_vec)
                        )
                    )
                    .where(NoteEmbedding.note_id == Note.id)
                    .scalar_subquery()
                )
                sem_filter = best_distance <= (1.0 - INTERACTIVE_SEARCH_THRESHOLD)
                query = query.where(sem_filter)
                count_query = count_query.where(sem_filter)
                semantic_order = best_distance.asc()
            else:
                terms = _strip_type_nouns(q)
                for term in terms:
                    escaped_term = term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
                    pattern = f"%{escaped_term}%"
                    term_filter = or_(Note.title.ilike(pattern), Note.body.ilike(pattern))
                    query = query.where(term_filter)
                    count_query = count_query.where(term_filter)

        if tags:
            for i, tag in enumerate(tags):
                param_tag = f"tag_{i}"
                param_prefix = f"tag_prefix_{i}"
                tag_filter = text(
                    f"EXISTS (SELECT 1 FROM unnest(notes.tags) AS t"
                    f" WHERE t = :{param_tag} OR t LIKE :{param_prefix})"
                ).bindparams(**{param_tag: tag, param_prefix: tag + "/%"})
                query = query.where(tag_filter)
                count_query = count_query.where(tag_filter)

        if status:
            statuses = [status] if isinstance(status, str) else status
            query = query.where(Note.status.in_(statuses))
            count_query = count_query.where(Note.status.in_(statuses))

        if priority:
            priorities = [priority] if isinstance(priority, str) else priority
            query = query.where(Note.priority.in_(priorities))
            count_query = count_query.where(Note.priority.in_(priorities))

        if due_before is not None:
            query = query.where(Note.due_date < due_before)
            count_query = count_query.where(Note.due_date < due_before)
        if due_after is not None:
            query = query.where(Note.due_date >= due_after)
            count_query = count_query.where(Note.due_date >= due_after)

        if project_id is not None:
            if milestone_ids:
                # OR: directly assigned to project, OR assigned to one of the project's milestones
                project_filter = or_(Note.project_id == project_id, Note.milestone_id.in_(milestone_ids))
            else:
                project_filter = Note.project_id == project_id
            query = query.where(project_filter)
            count_query = count_query.where(project_filter)

        if milestone_id is not None:
            query = query.where(Note.milestone_id == milestone_id)
            count_query = count_query.where(Note.milestone_id == milestone_id)

        if parent_id is not None:
            query = query.where(Note.parent_id == parent_id)
            count_query = count_query.where(Note.parent_id == parent_id)

        if task_kind is not None:
            query = query.where(Note.task_kind == task_kind)
            count_query = count_query.where(Note.task_kind == task_kind)

        if no_project:
            query = query.where(Note.project_id.is_(None))
            count_query = count_query.where(Note.project_id.is_(None))

        if exclude_paused_projects:
            from scribe.models.project import Project
            paused_ids = (
                select(Project.id)
                .where(Project.user_id == user_id, Project.status == "paused")
                .scalar_subquery()
            )
            paused_filter = or_(Note.project_id.is_(None), Note.project_id.not_in(paused_ids))
            query = query.where(paused_filter)
            count_query = count_query.where(paused_filter)

        if semantic_order is not None:
            # A query is a relevance claim — see the docstring.
            query = query.order_by(semantic_order)
        elif sort == "touched":
            # Last touched: the later of the task's own edit and its newest
            # work-log. Logging doesn't bump updated_at, so a task worked
            # through its logs would otherwise sort as untouched (#4045).
            # GREATEST skips a NULL, so a task with no logs sorts by its edit.
            from scribe.models.task_log import TaskLog
            last_log = (
                select(func.max(TaskLog.created_at))
                .where(TaskLog.task_id == Note.id)
                .scalar_subquery()
            )
            touched = func.greatest(Note.updated_at, last_log)
            query = query.order_by(touched.asc() if order == "asc" else touched.desc())
        else:
            sort_col = getattr(Note, sort, Note.updated_at)
            if order == "asc":
                query = query.order_by(sort_col.asc())
            else:
                query = query.order_by(sort_col.desc())

        query = query.limit(limit).offset(offset)

        total = await session.scalar(count_query) or 0
        result = await session.execute(query)
        notes = list(result.scalars().all())
        return notes, total


async def get_note_by_title(user_id: int, title: str) -> Note | None:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(
                Note.user_id == user_id,
                func.lower(Note.title) == func.lower(title.strip()),
                Note.deleted_at.is_(None),
            ).limit(1)
        )
        return result.scalars().first()


async def get_or_create_note_by_title(user_id: int, title: str) -> Note:
    title = title.strip()
    note = await get_note_by_title(user_id, title)
    if note:
        return note
    return await create_note(user_id, title=title)


# Kinds a caller may MINT. Narrower than what the COLUMN holds: `plan` is a
# valid stored value — historical plan-tasks carry it and must stay writable —
# but plans became milestones in 0066, so no door hands out a new one. The
# CHECK whitelist and this policy answer different questions, which is why
# they are deliberately not the same list.
MINTABLE_KINDS = ("work", "issue", "spike")


def minted_kind(kind: str) -> str:
    """Validate a kind a caller is asking to WRITE, or raise saying why.

    Lives here rather than in either door so both share one copy: the REST
    route cannot import an MCP tool module, and a second spelling of this
    list is how the two doors would come to disagree.

    Raises rather than falling back to 'work'. A silently-corrected kind is
    the defect this exists to end (#3129: the editor's Kind select reported
    success and changed nothing), and a caller naming a kind we do not know
    has a wrong idea that an error corrects and a default hides.
    """
    if kind in MINTABLE_KINDS:
        return kind
    if kind == "plan":
        raise ValueError(
            "kind='plan' is retired — plans are milestones. Call "
            "start_planning(project_id, title) to begin one. Existing "
            "plan-tasks keep the value and stay editable."
        )
    raise ValueError(f"kind must be one of {MINTABLE_KINDS}, got {kind!r}")


async def update_note(
    user_id: int, note_id: int, clear: Iterable[str] = (), **fields: object,
) -> Note | None:
    """Partial update. `clear` names fields to UNSET; **fields carries values.

    Clearing is explicit and separate because a nullable field cannot be
    emptied by passing it: the MCP door reads "" as "leave this alone", so an
    agent filling two fields does not wipe the others, and a note that stops
    being a constraint genuinely needs its check removed. Naming the field is
    the one form that cannot happen by accident. The REST door, where a
    cleared form input arrives as "", reaches the same place through the
    NULLABLE_NOTE_TEXT normalisation below — two idioms, one outcome.
    (Same shape as rulebooks.update_rule, milestone 312 step 2.)
    """
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(Note.id == note_id, Note.user_id == user_id)
        )
        note = result.scalars().first()
        if note is None:
            return None
        # Snapshot before changes for version creation
        old_body = note.body
        old_title = note.title
        old_tags = list(note.tags or [])
        check_before = note.verify_with
        for key in clear:
            if key in NULLABLE_NOTE_TEXT:
                setattr(note, key, None)
        for key, value in fields.items():
            if not hasattr(note, key):
                continue
            if key == "status" and isinstance(value, str):
                try:
                    value = TaskStatus(value).value
                except ValueError:
                    raise ValueError(f"Invalid status: {value!r}. Must be one of: {[s.value for s in TaskStatus]}")
            elif key == "priority" and isinstance(value, str):
                try:
                    value = TaskPriority(value).value
                except ValueError:
                    raise ValueError(f"Invalid priority: {value!r}. Must be one of: {[p.value for p in TaskPriority]}")
            elif key == "task_kind" and isinstance(value, str):
                # Same shape as status/priority above, and for the same
                # reason: a kind the column will refuse should fail here with
                # a readable message, not as a CheckViolationError from the
                # database. Before this, `task_kind` reached setattr through
                # the hasattr guard with no validation at all — but no door
                # ever offered it, so a task's kind was write-once (#3129).
                try:
                    value = TaskKind(value).value
                except ValueError:
                    raise ValueError(
                        f"Invalid kind: {value!r}. Must be one of: "
                        f"{[k.value for k in TaskKind]}"
                    )
            elif key == "tags" and isinstance(value, list):
                value = _normalize_tags(value)
            elif key in NULLABLE_NOTE_TEXT:
                value = value or None
            elif key == "verified_at":
                # Not settable here. A stamp says somebody performed THIS
                # check, so it is written by the verification path and by a
                # restore, never by an ordinary edit that could mint one for a
                # check nobody ran.
                continue
            setattr(note, key, value)
        # The invariant, over the RESULTING record rather than over what was
        # passed — which is what catches a checked note being turned into a
        # task. Raised before commit, so nothing is persisted.
        if note.verify_with or note.expires_when:
            guard_check_fields(note.status, note.note_type)
        # A stamp certifies A CHECK, not a record. Rewrite or remove the check
        # and the old stamp certifies something that no longer exists, so it is
        # dropped and the note re-enters the sweep. The safe direction: a note
        # wrongly listed as due costs one look; a note wrongly vouched for
        # costs exactly what the sweep exists to catch.
        if note.verify_with != check_before:
            note.verified_at = None
        # Some kinds derive `data` from their body — so a write that moves the
        # body through this generic door must move the mirror with it (#3128).
        # Without this, PATCH /api/notes/<id> {body} left the mirror behind,
        # and every read PREFERS the mirror: a snippet went on reporting its
        # old repo/path/symbol to prior-art recall while showing its new body,
        # and a lesson would go on being retrieved for the situation it used to
        # name. The kind's own updater composes the mirror itself and passes it
        # explicitly, so an explicit `data` always wins — the caller that knows
        # the field set beats the one that can only re-read the body.
        if "data" not in fields and not _PARSED_FROM_BODY.isdisjoint(fields):
            recompose = _mirror_recomposers().get(note.note_type or "")
            if recompose is not None:
                note.data = recompose(note)
        if "status" in fields:
            apply_status_transition(note, user_id)
        note.updated_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(note)

    # Create a version snapshot when body actually changes
    if "body" in fields and fields["body"] != old_body:
        from scribe.services.note_versions import create_version
        await create_version(user_id, note_id, old_body, old_title, old_tags)

    embed_note(note)

    if note.project_id is not None:
        await _maybe_reactivate_project(note.project_id)

    return note


# A hard `delete_note(user_id, note_id)` lived here with ZERO callers, and was
# removed with #278 step 1. It is recorded rather than silently dropped because
# the danger was never that it ran — it is that it was findable by name. Someone
# wanting to delete a note greps `delete_note`, finds a function in the notes
# service with exactly the right signature, and permanently destroys a record
# every path downstream expects to be recoverable.
#
# The delete path is `trash_svc.delete`, which soft-deletes an entity AND its
# descendants under one batch_id so `restore(batch)` works. `purge_trash` owns
# permanent deletion. Both are reachable; neither is spelled `delete_note`.


# ── The sweep (milestone 317 step 3) ─────────────────────────────────────────
#
# A SIBLING of rulebooks.rules_due_for_verification, not a shared
# implementation, and deliberately so (note 3163). The row could have been
# shared; the QUERY cannot. That sweep scopes by rulebook ownership XOR project
# ownership because rules have no sharing ACL at all — no rule_shares, no
# can_read_rule. A note scopes by the note ACL, which is a different question
# with a different answer. What IS common — how a stamp reads, how old it is —
# lives in services/verification.py and is imported by both.

async def notes_due_for_verification(
    user_id: int,
    older_than_days: int = 0,
    project_id: int | None = None,
    never_only: bool = False,
) -> list[Note]:
    """Notes that carry a check, oldest verification first, never-checked top.

    THE QUERY THE COLUMNS EXIST FOR. `verify_with` and `expires_when` are
    storage; this is what turns them into something that gets acted on.
    Without it, note decay is caught only when a human reads the note and
    disagrees — which is the case where the note was already believed.

    Ordered `verified_at` ASC **NULLS FIRST**: never-checked outranks
    checked-long-ago, because a note nobody has ever confirmed is a claim with
    no evidence behind it at all. Postgres sorts NULLs LAST on ASC by default,
    so this is explicit — and getting it wrong would not error, it would
    silently invert the one signal the sweep exists to carry.

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

    Scoped with `browsable_notes_clause`, NOT the read scope (decision note
    2094): a sweep is a passive surface, and a record shared one-to-one must
    not arrive in one unasked.

    Deliberately NOT filtered to non-task, non-snippet records even though the
    write path (step 2) permits a check on nothing else. A row in that state
    would be a row in an ILLEGAL state, and this is the one surface that could
    tell somebody about it. Hiding it here to match the invariant would make
    the sweep agree with a database it had stopped describing.

    Args:
        user_id: whose notes.
        older_than_days: only notes last verified longer ago than this.
            Never-checked notes always qualify — they are the most overdue
            thing there is. 0 = no age filter. Negative raises: it would mean
            "everything", which is a different question than the one asked,
            answered silently.
        project_id: narrow to one project. None = every project.
        never_only: only notes that have never been verified.
    """
    from datetime import timedelta

    from scribe.services.access import browsable_notes_clause

    if older_than_days < 0:
        raise ValueError(
            f"older_than_days must be >= 0, got {older_than_days}. A negative "
            f"window silently means 'everything', which is not what any caller "
            f"of a staleness sweep is asking."
        )

    async with async_session() as session:
        # One statement, not a fetch-then-filter: the ordering below is the
        # database's, so it cannot disagree with itself across two halves.
        stmt = (
            select(Note)
            .where(
                browsable_notes_clause(user_id),
                Note.deleted_at.is_(None),
                Note.verify_with.is_not(None),
            )
        )
        if project_id is not None:
            stmt = stmt.where(Note.project_id == project_id)
        if never_only:
            stmt = stmt.where(Note.verified_at.is_(None))
        elif older_than_days > 0:
            cutoff = datetime.now(timezone.utc) - timedelta(days=older_than_days)
            stmt = stmt.where(
                or_(Note.verified_at.is_(None), Note.verified_at < cutoff)
            )
        stmt = stmt.order_by(Note.verified_at.asc().nullsfirst(), Note.id)
        return list((await session.execute(stmt)).scalars().all())


def verification_row(note: Note) -> dict:
    """One row of the sweep — the CHECK in full, unlike a listing.

    The opposite call from a browse: here the caller is about to go and run the
    check, so the text they need IS the payload rather than the bloat.
    """
    from scribe.services.verification import (
        days_since_verified,
        last_verified_label,
    )

    return {
        "id": note.id,
        "title": note.title,
        "project_id": note.project_id,
        "verify_with": note.verify_with or "",
        "expires_when": note.expires_when or "",
        "last_verified": last_verified_label(note),
        "days_since_verified": days_since_verified(note),
    }


async def mark_note_verified(
    note_id: int, user_id: int, still_true: bool = True,
) -> Note | None:
    """Stamp a note as verified — or, when the check FAILED, refuse to.

    The asymmetry is the design: passing writes a stamp, failing writes
    nothing. There is no "verified false" state, because a note whose check
    failed is not a note in a special condition — it is a note that is WRONG,
    and the honest resolutions are to correct it, supersede it, or find out
    why. Recording the failure as a flag would let it sit there being false
    with the sweep quietly satisfied that somebody had looked.

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

    Write access, not read (rules 47/78): stamping is a mutation, and an
    editor-share holder may make it while a viewer may not.

    Returns None when the note is not found, not writable, or carries no
    `verify_with` — nothing to verify is a different answer from verified.
    """
    from scribe.services.access import can_write_note

    async with async_session() as session:
        note = (await session.execute(
            select(Note).where(Note.id == note_id, Note.deleted_at.is_(None))
        )).scalars().first()
        if note is None or not note.verify_with:
            return None
        if not await can_write_note(user_id, note_id):
            return None
        if still_true:
            note.verified_at = datetime.now(timezone.utc)
            await session.commit()
            await session.refresh(note)
        return note


async def get_all_tags(user_id: int, q: str | None = None) -> list[str]:
    async with async_session() as session:
        if q:
            result = await session.execute(
                text(
                    "SELECT DISTINCT tag FROM"
                    " (SELECT unnest(tags) AS tag FROM notes"
                    "  WHERE tags != '{}' AND user_id = :user_id) t"
                    " WHERE tag ILIKE :q_pattern ORDER BY tag LIMIT 100"
                ).bindparams(user_id=user_id, q_pattern=f"%{q}%")
            )
        else:
            result = await session.execute(
                text(
                    "SELECT DISTINCT unnest(tags) AS tag FROM notes"
                    " WHERE tags != '{}' AND user_id = :user_id"
                    " ORDER BY tag LIMIT 100"
                ).bindparams(user_id=user_id)
            )
        return [row[0] for row in result.fetchall()]


async def convert_note_to_task(user_id: int, note_id: int) -> Note:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(Note.id == note_id, Note.user_id == user_id)
        )
        note = result.scalars().first()
        if note is None:
            logger.warning("convert_note_to_task: note %d not found", note_id)
            raise ValueError("Note not found")
        note.status = TaskStatus.todo.value
        note.priority = TaskPriority.none.value
        note.updated_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(note)
        logger.info("Converted note %d to task", note_id)
        return note


async def convert_task_to_note(user_id: int, note_id: int) -> Note:
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(Note.id == note_id, Note.user_id == user_id)
        )
        note = result.scalars().first()
        if note is None:
            logger.warning("convert_task_to_note: note %d not found", note_id)
            raise ValueError("Note not found")
        note.status = None
        note.priority = None
        note.due_date = None
        # A plain note is not a task and must not recur — clear the rule and
        # any armed spawn timestamp so the recurrence sweep never picks it up.
        note.recurrence_rule = None
        note.recurrence_next_spawn_at = None
        note.updated_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(note)
        logger.info("Converted task %d to note", note_id)
        return note


async def resolve_process(user_id: int, name_or_id) -> tuple[Note | None, list[dict]]:
    """Resolve a stored process by id or name.

    note_type='process', non-trashed, scoped to what this user may READ — owned
    plus shared (rule #78). Naming one is an explicit act, so a Process shared
    with the caller resolves here even though it is deliberately absent from the
    passive process list and the skill manifest (decision note 2094); without
    this, a Process a search surfaced could not then be run — see #2093.

    Precedence: numeric id → exact case-insensitive title → substring. Returns
    (note, other_candidates); on a substring tie with no exact hit, `note` is the
    most-recently-updated match and `other_candidates` lists the rest as
    [{id, title}] so the caller can disambiguate. Returns (None, []) when nothing
    matches.
    """
    from scribe.services.access import readable_notes_clause
    visible = readable_notes_clause(user_id)
    async with async_session() as session:
        base = select(Note).where(
            visible,
            Note.note_type == "process",
            Note.deleted_at.is_(None),
        )
        s = str(name_or_id).strip()
        if s.isdigit():
            row = (await session.execute(base.where(Note.id == int(s)))).scalars().first()
            if row is not None:
                return row, []
        exact = (await session.execute(
            base.where(func.lower(Note.title) == s.lower()).order_by(Note.updated_at.desc())
        )).scalars().first()
        if exact is not None:
            return exact, []
        matches = (await session.execute(
            base.where(Note.title.ilike(f"%{s}%")).order_by(Note.updated_at.desc())
        )).scalars().all()
        if not matches:
            return None, []
        return matches[0], [{"id": n.id, "title": n.title} for n in matches[1:]]


async def get_notes_by_ids(user_id: int, note_ids: list[int]) -> dict[int, Note]:
    """Batch fetch notes by ID list. Returns {note_id: Note}."""
    if not note_ids:
        return {}
    async with async_session() as session:
        result = await session.execute(
            select(Note).where(
                Note.user_id == user_id, Note.id.in_(note_ids), Note.deleted_at.is_(None)
            )
        )
        return {n.id: n for n in result.scalars().all()}


async def get_backlinks(user_id: int, note_id: int) -> list[dict]:
    note = await get_note(user_id, note_id)
    if note is None:
        return []

    title = note.title
    if not title:
        return []

    async with async_session() as session:
        escaped = title.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        pattern = f"%[[{escaped}]]%"
        pattern_alias = f"%[[{escaped}|%"

        results = await session.execute(
            select(Note.id, Note.title, Note.status).where(
                Note.user_id == user_id,
                Note.id != note_id,
                Note.deleted_at.is_(None),
                or_(Note.body.like(pattern), Note.body.like(pattern_alias)),
            )
        )
        backlinks: list[dict] = []
        for row in results.fetchall():
            link_type = "task" if row[2] is not None else "note"
            backlinks.append({"type": link_type, "id": row[0], "title": row[1]})

        return backlinks


_WIKILINK_RE = re.compile(r'\[\[([^\]|]+)(?:\|[^\]]+)?\]\]')


async def build_note_graph(
    user_id: int,
    project_id: int | None = None,
    include_shared_tags: bool = False,
) -> dict:
    notes, _ = await list_notes(user_id, project_id=project_id, limit=1000)
    if not notes:
        return {"nodes": [], "edges": []}

    # Fetch project colours
    project_ids = {n.project_id for n in notes if n.project_id is not None}
    project_colors: dict[int, str] = {}
    if project_ids:
        from scribe.models.project import Project
        async with async_session() as session:
            result = await session.execute(
                select(Project.id, Project.color).where(Project.id.in_(project_ids))
            )
            for pid, color in result.fetchall():
                project_colors[pid] = color or "#888888"

    # Build title lookup (lowercase → id)
    title_map: dict[str, int] = {}
    for n in notes:
        if n.title:
            title_map[n.title.lower()] = n.id

    # Build nodes
    nodes = []
    for n in notes:
        nodes.append({
            "id": n.id,
            "title": n.title or "(untitled)",
            "type": "task" if n.is_task else "note",
            "tags": n.tags or [],
            "project_id": n.project_id,
            "project_color": project_colors.get(n.project_id) if n.project_id else None,
        })

    # Build wikilink edges
    edge_set: set[tuple[int, int]] = set()
    edges = []
    for n in notes:
        if not n.body:
            continue
        for match in _WIKILINK_RE.findall(n.body):
            target_id = title_map.get(match.strip().lower())
            if target_id is not None and target_id != n.id:
                pair = (n.id, target_id)
                if pair not in edge_set:
                    edge_set.add(pair)
                    edges.append({"source": n.id, "target": target_id, "type": "wikilink"})

    # Build tag nodes + note→tag edges
    if include_shared_tags:
        tag_to_ids: dict[str, list[int]] = {}
        for n in notes:
            for tag in (n.tags or []):
                tag_to_ids.setdefault(tag, []).append(n.id)

        for tag, note_ids in tag_to_ids.items():
            tag_node_id = f"tag:{tag}"
            nodes.append({
                "id": tag_node_id,
                "title": tag,
                "type": "tag",
                "tags": [],
                "project_id": None,
                "project_color": None,
            })
            for nid in note_ids:
                edges.append({"source": nid, "target": tag_node_id, "type": "tag"})

    return {"nodes": nodes, "edges": edges}


# ---------------------------------------------------------------------------
# Shared-access variant
# ---------------------------------------------------------------------------

async def get_note_for_user(
    accessing_user_id: int, note_id: int
) -> tuple["Note", str] | None:
    """Returns (note, permission) if user has any access, else None."""
    from scribe.services.access import get_note_permission
    perm = await get_note_permission(accessing_user_id, note_id)
    if perm is None:
        return None
    async with async_session() as session:
        note = await session.get(Note, note_id)
    return (note, perm) if note else None


# What a LISTING row carries (#4061): what the record is and where it sits,
# never what it says. list_tasks once returned every row's to_dict(), body
# included, and a project's todo list came to 93-165k characters, past what an
# MCP client accepts inline, so the list arrived as a file to page through. The
# same failure #4045 fixed for enter_project, one call over. get_task / get_note
# read a record in full; a list is for choosing which one to open.
#
# A field most rows leave empty (a one-line description, a parent, a due date)
# is attached only when set: a hundred rows of `null` are a hundred chances to
# learn to skip the key (#2483), and the bytes are the thing being cut.
def brief_row(
    note: Note,
    milestone_titles: dict[int, str] | None = None,
    log_counts: dict[int, int] | None = None,
) -> dict:
    """One task as a list row — what it is and where it stands, not its body.

    `log_counts` is a whole page's work-log counts fetched in one aggregate
    (task_logs.log_counts_for_tasks). Passed in rather than looked up here so
    the N+1 stays impossible, and applied to tasks only: a note has no work
    log. Zero-filled when the mapping is given, because a row that omits the
    key says "no record" on every row rather than on the ones with none, and
    knowing WHICH rows carry a record is how a reader decides what to open.
    """
    row = {
        "id": note.id,
        "title": note.title,
        "note_type": note.note_type or "note",
        "project_id": note.project_id,
        "tags": note.tags or [],
        "updated_at": iso(note.updated_at),
    }
    if note.description:
        row["description"] = note.description
    if note.is_task:
        row.update({
            "status": note.status,
            "task_kind": note.task_kind,
            "priority": note.priority,
            "milestone_id": note.milestone_id,
        })
        if milestone_titles is not None and note.milestone_id:
            row["milestone_title"] = milestone_titles.get(note.milestone_id)
        if note.parent_id:
            row["parent_id"] = note.parent_id
        if note.due_date:
            row["due_date"] = iso(note.due_date)
        if log_counts is not None:
            row["log_count"] = log_counts.get(int(note.id), 0)
    return row