import logging
import re
from collections.abc import 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

logger = logging.getLogger(__name__)

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


# 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.
    """
    try:
        import asyncio

        from scribe.services.embeddings import upsert_note_embedding
        # Chunking and the empty-record gate live inside upsert_note_embedding —
        # one path for every writer (#280).
        asyncio.create_task(
            upsert_note_embedding(note.id, note.user_id, note.title, 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:
    # 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]}")

    # 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

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


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.
    """
    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)
        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
        # A snippet's `data` is DERIVED from its body — so a write that moves
        # the body through this generic door must move the mirror with it
        # (#3128). Without this, PATCH /api/notes/<snippet_id> {body} left the
        # mirror behind, and snippet_fields PREFERS the mirror: the row went on
        # reporting its old repo/path/symbol to prior-art recall while showing
        # its new body. `update_snippet` 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):
            # Imported here, not at module scope: services/snippets.py calls
            # back into this module (update_snippet -> update_note), so a
            # top-level import is a cycle.
            from scribe.services.snippets import (
                SNIPPET_NOTE_TYPE, recompose_data,
            )
            if note.note_type == SNIPPET_NOTE_TYPE:
                note.data = recompose_data(note)
        # Auto-set lifecycle timestamps on status transitions
        if "status" in fields:
            _now = datetime.now(timezone.utc)
            if note.status == TaskStatus.in_progress.value:
                if note.started_at is None:
                    note.started_at = _now
            elif note.status in (TaskStatus.done.value, TaskStatus.cancelled.value):
                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:
                note.started_at = None
                note.completed_at = None
                note.recurrence_next_spawn_at = None
        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