"""Session-context rendering for the Scribe plugin's SessionStart hook.

The plugin's hook curls `GET /api/plugin/context` at session start and injects
the returned text as `additionalContext`, giving Scribe the same push channel
that superpowers and file-memory have. This module renders that text.

Design note — altitude: we inject rule *titles* grouped by topic (a compact
index), NOT every rule's full statement. The 48 always-on statements run well
past the 10k-char `additionalContext` cap, and the push channel's job is to make
Claude *aware* the rules exist and *reach* for them — not to dump them. Full
text stays one `get_rule(id)` / `list_always_on_rules()` call away. Titles are
mostly self-describing ("`dev` is home", "No GitHub — Fabled-Git only"), so the
index alone already steers behavior.
"""
from __future__ import annotations

import logging
import re
import time

from sqlalchemy import select

from scribe.models import async_session
from scribe.models.rulebook import RulebookTopic
from scribe.services import design_systems as design_systems_svc
from scribe.services import knowledge as knowledge_svc
from scribe.services import notes as notes_svc
from scribe.services import projects as projects_svc
from scribe.services import rulebooks as rulebooks_svc
from scribe.services import shape_ledger as shape_ledger_svc
from scribe.services import snippets as snippets_svc
from scribe.services.access import label_shared_items, owner_names_for
from scribe.services.embeddings import semantic_search_notes
from scribe.services.note_usage import record_surfaced
from scribe.services.supersession import superseded_ids
from scribe.services.retrieval_telemetry import record_retrieval
from scribe.services.settings import get_setting

logger = logging.getLogger(__name__)

# Defensive cap below Claude Code's 10k additionalContext limit.
_MAX_CHARS = 9000

# Max chars of a Process body to fold into the auto-surface description.
_PROC_PREVIEW_CHARS = 200

# --- Knowledge auto-inject (Path A: per-turn awareness push) -----------------
# Per-user settings (keys live in the generic settings table). The threshold is
# deliberately STRICTER than the pull-search default (embeddings
# DEFAULT_SIMILARITY_THRESHOLD = 0.45): an unsolicited per-turn inject must clear
# a higher bar than a search the agent chose to run. Defaults start conservative
# and are meant to be tuned from retrieval_logs (source='auto_inject') once data
# accrues — they're exposed in the Settings UI, no restart needed.
AUTOINJECT_ENABLED_KEY = "kb_autoinject_enabled"
AUTOINJECT_THRESHOLD_KEY = "kb_autoinject_threshold"
AUTOINJECT_TOP_K_KEY = "kb_autoinject_top_k"

AUTOINJECT_DEFAULT_ENABLED = True
AUTOINJECT_DEFAULT_THRESHOLD = 0.55
AUTOINJECT_DEFAULT_TOP_K = 3

# The write-path trigger (#2082) gets its own on/off switch, its own threshold,
# and shares only top-k. It originally shared the threshold too, on the argument
# that one "how loud may Scribe be" knob beats two that drift — and reserved the
# split for when telemetry showed the two surfaces wanted different values.
#
# #2223 is that evidence. Measured against the live instance, the semantic arm's
# scores for CODE sit far above what the same threshold means for PROSE:
#   near-duplicate of a recorded helper   0.73-0.74   (true positive)
#   unrelated colour math / Vue SFC / CSS 0.55-0.63   (false positive)
#   `x = 1`                               0.58        (false positive)
# Any two Python-shaped payloads share keywords, indentation and structure, so
# the floor for "some code" is ~0.55-0.63 — auto-inject's 0.55 lands INSIDE that
# noise band, and 6 of 8 probe payloads produced a nudge (4 of them noise). The
# margin gate can't rescue it either: _AUTOINJECT_BAND is relative to the top
# hit, so with a single hit it never engages.
#
# 0.68 clears every measured false positive with margin and still sits 0.05
# below both true positives. Auto-inject keeps 0.55 — it was tuned on prose and
# is not implicated. Tune from retrieval_logs (source='write_path') + note_usage
# pull-through (#2085) once a real corpus accrues; a cross-encoder rerank
# (#1038) would subsume this bump.
WRITEPATH_ENABLED_KEY = "kb_writepath_enabled"
WRITEPATH_THRESHOLD_KEY = "kb_writepath_threshold"
WRITEPATH_DEFAULT_ENABLED = True
WRITEPATH_DEFAULT_THRESHOLD = 0.68

# Minimum SUBSTANCE (non-whitespace chars) a payload must carry before the
# semantic arm will run at all — the cheap half of the operator's #89 idea
# ("a sliding scale between number of characters and semantic threshold").
#
# Deliberately NOT a settings knob and deliberately conservative. Its job is
# only to drop payloads too small to carry meaning, where an embedding is noise
# rather than signal: `x = 1`, a renamed variable, a changed string literal —
# which is what most single-line Edits look like, and the majority of Edits are
# single-line. 48 sits below the smallest plausible reusable helper (a one-line
# `def` with a body runs ~60), so it errs toward keeping recall and leaves
# precision to the threshold above, which is where the measured separation is.
# The full length↔threshold CURVE is still open in #89 — the operator flagged it
# as wanting a brainstorm, so this stays a flat floor rather than an invented
# scale. It also saves a pointless embedding round-trip on trivial edits.
WRITEPATH_MIN_CODE_CHARS = 48

# --- concept extraction for the semantic arm's query (#2242) ------------------
# A snippet's embedded text is f"{title}\n{body}", and for a snippet that body is
# composed markdown: **When to use:**, **Signature:**, **Location:**, then the
# fenced code. So `when_to_use` — the description of what the thing is FOR —
# appears twice in the vector, and the document is prose-forward.
#
# The arm used to query it with raw code and no prose at all. Measured on the
# deployed instance against snippet #2222, same corpus:
#   query built from            score   best unrelated   separation
#   raw code body               0.743   0.630            0.11
#   name + docstring            0.823   0.602            0.22
#   hand-written concept prose  0.835   0.583            0.25
# A 12-word description beats a near-verbatim reimplementation of the function,
# and code-as-query RAISES the noise floor. It is also the cleanest explanation
# for the fragment miss recorded on #2223: a short code excerpt has almost no
# prose to match against a document that is mostly prose.
#
# So we send the concept instead — and shape it like a snippet's own title,
# "{name} — {when_to_use}", because that is the form the 0.823 measurement used.
# Undocumented code yields little, and a Vue SFC or a config file yields nothing;
# those fall back to the raw payload and behave exactly as before. This raises
# the ceiling for documented helpers rather than fixing every case.

# Declaration forms, one pattern per shape, every pattern exposing (name, params)
# so composition doesn't have to care which matched. Deliberately regex and not a
# real parser: this runs on a PreToolUse hook's critical path, the payload is
# frequently a FRAGMENT that no parser would accept (an Edit's new_string is
# rarely a valid module), and a miss costs only a fallback to today's behaviour.
_CONCEPT_DECL_PATTERNS = (
    # python: def / async def, and class with optional bases
    re.compile(r"^[ \t]*(?:async[ \t]+)?def[ \t]+([A-Za-z_]\w*)[ \t]*(\([^)]*\))", re.M),
    re.compile(r"^[ \t]*class[ \t]+([A-Za-z_]\w*)[ \t]*(\([^)]*\))?", re.M),
    # js/ts: function decl, and the const-arrow form that dominates modern code
    re.compile(r"^[ \t]*(?:export[ \t]+)?(?:default[ \t]+)?(?:async[ \t]+)?function[ \t]+([A-Za-z_$][\w$]*)[ \t]*(\([^)]*\))", re.M),
    re.compile(r"^[ \t]*(?:export[ \t]+)?(?:const|let|var)[ \t]+([A-Za-z_$][\w$]*)[ \t]*=[ \t]*(?:async[ \t]*)?(\([^)]*\))[ \t]*=>", re.M),
    # rust / go
    re.compile(r"^[ \t]*(?:pub[ \t]+)?fn[ \t]+([A-Za-z_]\w*)[ \t]*(\([^)]*\))", re.M),
    re.compile(r"^[ \t]*func[ \t]+(?:\([^)]*\)[ \t]*)?([A-Za-z_]\w*)[ \t]*(\([^)]*\))", re.M),
    # posix shell: name() {
    re.compile(r"^[ \t]*([A-Za-z_]\w*)[ \t]*(\(\))[ \t]*\{", re.M),
)

# Doc forms, tried in order. The Python pattern also matches a triple-quoted
# string that isn't a docstring — accepted: a stray literal is still text about
# what the code does far more often than it's misleading, and the cost is a
# slightly worse query rather than a wrong answer.
_CONCEPT_PY_DOC = re.compile(r'("""|\'\'\')(.*?)\1', re.S)
_CONCEPT_JSDOC = re.compile(r"/\*\*(.*?)\*/", re.S)
_CONCEPT_LEADING_COMMENT = re.compile(r"\A(?:[ \t]*(?://|#)[^\n]*\n?)+")
# A shebang is a comment to the regex above but says nothing about what the code
# DOES, and it would otherwise open the doc with "/usr/bin/env bash".
_CONCEPT_SHEBANG = re.compile(r"\A#![^\n]*\n")
_CONCEPT_COMMENT_MARKER = re.compile(r"^[ \t]*(?://+|#+!?)[ \t]?", re.M)
_CONCEPT_JSDOC_STAR = re.compile(r"^[ \t]*\*+[ \t]?", re.M)

# Cap the doc so a long module docstring can't drown out the declaration, and cap
# declarations so a 40-function Write doesn't turn into a wall of signatures.
_CONCEPT_MAX_DOC_CHARS = 400
_CONCEPT_MAX_DECLS = 4
# Below this much substance the "concept" is too thin to be a better query than
# the code itself (e.g. all we found was `f()`), so we keep the raw payload.
_CONCEPT_MIN_CHARS = 16

# Margin gate: drop any hit more than this far below the top hit's score, so a
# single strong match doesn't drag in a wall of barely-passing neighbours.
_AUTOINJECT_BAND = 0.10
# Hard ceiling on top-k regardless of the user's setting — this is an
# awareness menu (titles only), never a content dump.
_AUTOINJECT_MAX_TOP_K = 10


def _slugify(text: str) -> str:
    """kebab-case slug for a skill directory name (a-z0-9 + single hyphens)."""
    s = re.sub(r"[^a-z0-9]+", "-", (text or "").lower()).strip("-")
    return s or "process"


async def build_process_manifest(user_id: int) -> dict:
    """List the user's stored Processes as auto-surfacing skill-stub specs.

    The plugin's sync script (scribe_sync_processes.sh) writes one
    ~/.claude/skills/scribe-proc-<slug>/SKILL.md per entry — `description` is the
    auto-surface trigger, and the stub body calls get_process(name) for the live
    procedure (single source of truth in the DB). Reuses the list_processes query
    (note_type='process'). Instance-agnostic: derived from whatever Processes the
    calling install owns, no operator-specific coupling.

    SCOPE: this is the most consequential passive surface Scribe has — every
    entry becomes a skill file on the operator's machine that auto-surfaces and
    is followed as written. It therefore uses the BROWSE scope (via the
    no-query knowledge list): a Process shared directly with the operator is
    never installed here, only one they own or reach through a shared project
    (decision note 2094). Project-shared entries are labelled with their owner so
    the stub can't pass off someone else's procedure as the operator's own.

    Returns {"processes": [{id, name, slug, description, shared?, owner?}],
    "total": int}. Slugs are unique within the result (collision gets -<id>).
    """
    items, _ = await knowledge_svc.query_knowledge(
        user_id=user_id, note_type="process", tags=[], sort="modified",
        q=None, limit=100, offset=0,
    )
    items = await label_shared_items(user_id, items)
    procs: list[dict] = []
    seen: set[str] = set()
    for it in items:
        title = (it.get("title") or "").strip()
        if not title:
            continue
        slug = _slugify(title)
        if slug in seen:
            slug = f"{slug}-{it['id']}"
        seen.add(slug)

        preview = " ".join((it.get("snippet") or "").split())
        if len(preview) > _PROC_PREVIEW_CHARS:
            preview = preview[:_PROC_PREVIEW_CHARS].rstrip() + "…"
        if it.get("shared"):
            owner = it.get("owner") or "another user"
            description = (
                f'A shared Scribe process "{title}", authored by {owner} — NOT the'
                f" operator's own."
                + (f" {preview}" if preview else "")
                + f' Use only when the operator asks to run the "{title}" process'
                f" by name — and even then summarise it and get their go-ahead"
                f" first, since it reflects {owner}'s judgement rather than"
                f" theirs. If a request merely resembles this process, the live"
                f" instructions govern: offer it by name, don't follow it."
            )
        else:
            description = (
                f'Run the operator\'s saved Scribe process "{title}".'
                + (f" {preview}" if preview else "")
                + f' Use when the operator asks to run the "{title}" process by'
                f" name. If a request merely RESEMBLES this process, the live"
                f" instructions govern — offer the process by name and ask"
                f" before following it; never substitute it for explicit"
                f" instructions, and never inherit approvals embedded in it"
                f" (e.g. a fan-out opt-in) the operator hasn't granted in this"
                f" conversation. When you do run it, the process is the"
                f" skeleton and the conversation supplies the parameters:"
                f" constraints stated live override its defaults, and clarify"
                f" questions the conversation already answers are confirmed,"
                f" not re-asked."
            )
        entry = {
            "id": it["id"], "name": title, "slug": slug,
            "description": description,
        }
        if it.get("shared"):
            entry["shared"] = True
            entry["owner"] = it.get("owner")
        procs.append(entry)

    # The most consequential passive surface Scribe has (see SCOPE above), and
    # it emitted nothing — a Process installed as a skill, matched on every
    # relevant turn and never once opened, was indistinguishable from one never
    # installed (#2477). The honest event is "installed on the operator's
    # machine", which is a surfacing in effect: the skill description is in
    # front of the model each session. AMBIENT source — installation is not a
    # ranked choice — so it lands in ambient_count, not surfaced_count.
    record_surfaced(
        user_id=user_id,
        note_ids=[int(p["id"]) for p in procs],
        source="process_skill_sync",
    )
    return {"processes": procs, "total": len(procs)}


async def get_autoinject_config(user_id: int) -> dict:
    """Resolve a user's auto-inject settings, falling back to the defaults.

    Returns {"enabled": bool, "threshold": float, "top_k": int}, clamped to
    sane ranges (threshold to [0,1]; top_k to [1, _AUTOINJECT_MAX_TOP_K]).
    """
    enabled_raw = await get_setting(
        user_id, AUTOINJECT_ENABLED_KEY,
        "true" if AUTOINJECT_DEFAULT_ENABLED else "false",
    )
    enabled = enabled_raw.strip().lower() in ("true", "1", "yes", "on")

    try:
        threshold = float(await get_setting(
            user_id, AUTOINJECT_THRESHOLD_KEY, str(AUTOINJECT_DEFAULT_THRESHOLD)))
    except (TypeError, ValueError):
        threshold = AUTOINJECT_DEFAULT_THRESHOLD
    threshold = min(1.0, max(0.0, threshold))

    try:
        top_k = int(float(await get_setting(
            user_id, AUTOINJECT_TOP_K_KEY, str(AUTOINJECT_DEFAULT_TOP_K))))
    except (TypeError, ValueError):
        top_k = AUTOINJECT_DEFAULT_TOP_K
    top_k = min(_AUTOINJECT_MAX_TOP_K, max(1, top_k))

    return {"enabled": enabled, "threshold": threshold, "top_k": top_k}


def _record_kind(note) -> str:
    """The one-word kind marker for an injected menu line.

    The menu is drawn from every record that carries an embedding, so a snippet,
    a stored process, an issue and a stray dev-log all arrive looking identical.
    Recorded prior art only stands out if the line says what it is — and the kind
    is also what tells the reader which tool opens it.

    Task-ness wins over `note_type` because it's the more useful distinction at a
    glance: "there's an open issue about this" beats "there's a note about this".
    """
    if note.is_task:
        return "issue" if note.task_kind == "issue" else "task"
    return note.note_type or "note"


_REUSE_KINDS = ("snippet", "process")


async def _reserve_slot_for_reuse(
    user_id: int,
    query: str,
    kept: list,
    cfg: dict,
    *,
    project_id: int | None,
    exclude_ids: set[int],
) -> list:
    """Guarantee the reuse-shaped kinds one slot, if one clears threshold (#2246).

    Ranking by raw cosine is blind to what KIND of record answers what kind of
    ask, and the corpus makes that fatal rather than merely imperfect: Scribe's
    project records are *about software work*, so a task titled "surface snippets
    before the agent writes code" is a near-perfect lexical match for "write a
    function…" while being useless as an answer to it. Measured live, a prompt
    asking for a helper returned three records about BUILDING the retrieval
    system and zero snippets.

    The bias is structural and gets WORSE as the project record grows — which is
    the direction Scribe is supposed to grow. Snippets are ~0.5% of the corpus
    here; no threshold tuning fixes a 200:1 ratio.

    So the reserved hit is deliberately NOT held to the margin band. The band
    measures distance from the top overall score, and that top score is the very
    thing snippets lose to. It still has to clear the configured threshold, so a
    weak snippet cannot buy the slot — silence stays the default.
    """
    if any(_record_kind(n) in _REUSE_KINDS for _s, n in kept):
        return kept                      # reuse already represented; nothing to do

    top_k = cfg["top_k"]
    _t0 = time.perf_counter()
    reuse = await semantic_search_notes(
        user_id, query,
        limit=1,
        threshold=cfg["threshold"],
        project_id=project_id,
        exclude_ids=exclude_ids | {int(n.id) for _s, n in kept},
        note_type=_REUSE_KINDS,
        scope="browse",
    )
    # A real semantic query competing for a menu slot — logged like the scored
    # arm it displaces. Before this, the hit it PUSHED OUT was in
    # retrieval_logs and the query that pushed it out was not, so the slot
    # could never be evaluated against what it replaced (#2463; #1038 and
    # #2085 are gated on this ledger being complete).
    record_retrieval(
        user_id=user_id, source="reuse_slot", query=query,
        threshold=cfg["threshold"], limit=1, project_id=project_id,
        is_task=None, results=reuse,
        duration_ms=(time.perf_counter() - _t0) * 1000.0,
    )
    # Verify the kind rather than trusting the query that asked for it, and
    # dedup on top of exclude_ids. This slot exists FOR reuse kinds — a slot
    # silently spent on something else is worse than no slot, because the line
    # is indistinguishable from one that earned its place on score.
    kept_ids = {int(n.id) for _s, n in kept}
    fresh = [
        (s, n) for s, n in reuse
        if _record_kind(n) in _REUSE_KINDS and int(n.id) not in kept_ids
    ][:1]
    if not fresh:
        return kept

    # Take the LAST slot, never the first: the strongest overall hit is still the
    # best answer to the prompt, and displacing it would trade one blindness for
    # another.
    if len(kept) >= top_k:
        return kept[:top_k - 1] + fresh
    return (kept + fresh)[:top_k]


async def build_autoinject_hint(
    user_id: int,
    query: str,
    project_id: int = 0,
    exclude_ids: list[int] | None = None,
) -> dict:
    """Title-first awareness hint for the plugin's UserPromptSubmit hook.

    The four anti-bloat gates (see the module + milestone-93 design):
      1. high-confidence threshold (stricter than pull) — set per-user;
      2. margin gate — keep only hits within _AUTOINJECT_BAND of the top score;
      3. session dedup — caller passes already-injected ids as `exclude_ids`;
      4. title-first payload — id + kind + title + score only, never bodies.
    Disabled, blank-query, or nothing-clears-the-gates all return empty context,
    so most turns inject nothing.

    Returns {"context": str, "note_ids": list[int], "config": dict}. Every
    retrieval (even empty) is logged to retrieval_logs as source='auto_inject'
    so the threshold can be tuned from data.
    """
    cfg = await get_autoinject_config(user_id)
    empty = {"context": "", "note_ids": [], "config": cfg}
    q = (query or "").strip()
    if not cfg["enabled"] or not q:
        return empty

    t0 = time.perf_counter()
    hits = await semantic_search_notes(
        user_id, q,
        limit=cfg["top_k"],
        threshold=cfg["threshold"],
        project_id=(project_id or None),
        exclude_ids=set(exclude_ids or []),
        # Injection is the one retrieval nobody asked for, so it takes the BROWSE
        # scope: never a record shared one-to-one with the operator. What can
        # still appear is a collaborator's note inside a shared project — legible
        # only because the line below names its owner.
        scope="browse",
    )
    record_retrieval(
        user_id=user_id, source="auto_inject", query=q,
        threshold=cfg["threshold"], limit=cfg["top_k"],
        project_id=(project_id or None), is_task=None, results=hits,
        duration_ms=(time.perf_counter() - t0) * 1000.0,
    )
    if not hits:
        return empty

    # Margin gate: keep only hits close to the strongest one.
    top_score = hits[0][0]
    kept = [(s, n) for s, n in hits if s >= top_score - _AUTOINJECT_BAND]
    kept = await _reserve_slot_for_reuse(
        user_id, q, kept, cfg, project_id=(project_id or None),
        exclude_ids=set(exclude_ids or []),
    )

    # A collaborator's note can reach this menu via a shared project, and the
    # operator never asked for it — so say whose it is. Unattributed, it reads as
    # something they wrote and settled.
    owners = await owner_names_for({
        int(n.user_id) for _s, n in kept if n.user_id != user_id
    })

    # "records", not "notes" — the menu can hold snippets, processes and tasks
    # too, and the kind marker on each line is only legible if the header doesn't
    # already claim they're all one thing.
    lines = [
        "> Possibly relevant from your Scribe records — open any in full with "
        "`get_note(id)`, or `get_snippet` / `get_process` for those kinds "
        "(titles only; injected once per session):",
    ]
    # A superseded record is DEMOTED, not removed (#278) — so one can still reach
    # this menu, and when it does the reader has to be told. An agent handed
    # stale material with nothing marking it acts on it with full confidence,
    # which is worse than never having surfaced it. One query for the whole menu.
    stale = await superseded_ids([int(n.id) for _s, n in kept])

    note_ids: list[int] = []
    for score, note in kept:
        note_ids.append(int(note.id))
        title = (note.title or "(untitled)").replace("\n", " ").strip()
        line = f"> - #{note.id} [{_record_kind(note)}] \"{title}\" ({score:.2f})"
        if int(note.id) in stale:
            line += " — SUPERSEDED, a later record covers this; check that first"
        if note.user_id != user_id:
            who = owners.get(int(note.user_id)) or "another user"
            line += f" — shared by {who}, treat as a suggestion"
        lines.append(line)

    # Records what SURVIVED the margin gate, not what the ranker returned — the
    # menu the agent actually saw. retrieval_logs already holds the full
    # candidate set for threshold tuning; conflating the two would make
    # "surfaced" mean two different things depending on the surface (#2085).
    record_surfaced(user_id=user_id, note_ids=note_ids, source="auto_inject")

    return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}


# --- Write-path trigger (#2082): prior art at the moment code is written ------
# Auto-inject above fires on the operator's prompt. The moment reuse is actually
# lost is later — when the AGENT decides mid-task to write a helper — and nothing
# fired there. This is that trigger: the plugin's PreToolUse hook on Write/Edit
# asks what prior art is already recorded for the file being written.
#
# Two arms, deliberately different in kind:
#   - BY PLACE — a snippet recorded at this path (or in its directory) is prior
#     art by definition, not by resemblance, so it isn't scored or thresholded.
#     This is what the reverse lookup (#2083) was built to answer.
#   - BY MEANING — semantic search over snippets only, using the code about to be
#     written, under the same gates as auto-inject.
# Place beats meaning in the menu because "there is already a canonical helper in
# this exact file" is a stronger claim than "this resembles something".


def _prior_art_line(item: dict, marker: str, owner: str | None, foreign_lang: str = "") -> str:
    """One menu line: `- #12 [here] "title"`, attributed when it isn't yours.

    A foreign language is folded into the marker (`[similar 0.72 · python]`)
    rather than appended after the title, so the reader sees it while still
    reading the score — the two together are the judgement being offered.
    """
    title = (item.get("title") or "(untitled)").replace("\n", " ").strip()
    mark = f"{marker} · {foreign_lang}" if foreign_lang else marker
    line = f"> - #{item['id']} [{mark}] \"{title}\""
    if owner:
        line += f" — shared by {owner}, treat as a suggestion"
    return line


# --- cross-language prior art (#2244) ----------------------------------------
# Retrieval is concept-shaped now, and concepts are language-agnostic: a query
# about a TypeScript union-find matches a PYTHON snippet at 0.72-0.73, comfortably
# over the bar. That is a feature — the operator's framing is "borrow the shape of
# the solution even when the code isn't directly reusable" — but only if the line
# SAYS so. An unlabelled Python hit offered while writing TypeScript either gets
# dismissed as irrelevant or, worse, pasted into a .ts file. Measured note: this
# cross-language matching predates concept queries; it was always happening, just
# never disclosed.
#
# Deliberately NOT gated behind a stricter threshold for foreign-language hits: a
# higher bar would suppress exactly the shape-borrowing this is for. Label, don't
# filter.
_LANG_BY_EXT = {
    "py": "python", "pyi": "python",
    "ts": "typescript", "tsx": "typescript", "mts": "typescript", "cts": "typescript",
    "js": "javascript", "jsx": "javascript", "mjs": "javascript", "cjs": "javascript",
    "vue": "vue", "svelte": "svelte",
    "go": "go", "rs": "rust", "rb": "ruby", "php": "php",
    "java": "java", "kt": "kotlin", "kts": "kotlin", "scala": "scala",
    "c": "c", "h": "c", "cc": "cpp", "cpp": "cpp", "cxx": "cpp", "hpp": "cpp",
    "cs": "csharp", "swift": "swift", "m": "objectivec", "mm": "objectivec",
    "sh": "shell", "bash": "shell", "zsh": "shell", "fish": "shell",
    "sql": "sql", "css": "css", "scss": "scss", "less": "less",
    "html": "html", "htm": "html", "yml": "yaml", "yaml": "yaml",
    "toml": "toml", "ini": "ini", "dockerfile": "dockerfile",
    "ex": "elixir", "exs": "elixir", "erl": "erlang", "hs": "haskell",
    "lua": "lua", "pl": "perl", "r": "r", "dart": "dart", "zig": "zig",
}

# `language` on a snippet is operator-typed free text, so fold the spellings that
# mean the same thing before comparing. Anything unrecognised passes through
# lowercased — an unknown-but-equal pair still compares equal, which is the only
# thing this needs to get right.
_LANG_ALIASES = {
    "py": "python", "python3": "python",
    "ts": "typescript", "tsx": "typescript",
    "js": "javascript", "jsx": "javascript", "node": "javascript",
    "sh": "shell", "bash": "shell", "zsh": "shell", "shell-script": "shell",
    "c++": "cpp", "cplusplus": "cpp", "c#": "csharp", "objective-c": "objectivec",
    "golang": "go", "rs": "rust", "rb": "ruby", "yml": "yaml",
    "postgres": "sql", "postgresql": "sql", "psql": "sql",
    "vuejs": "vue", "vue3": "vue",
}


def _canonical_language(name: str) -> str:
    """Fold a free-text language name to a comparable token ("" if absent)."""
    token = (name or "").strip().lower()
    return _LANG_ALIASES.get(token, token)


def _language_for_path(path: str) -> str:
    """The language implied by a file path's extension ("" when unknown)."""
    tail = (path or "").rsplit("/", 1)[-1].lower()
    if tail.startswith("dockerfile"):
        return "dockerfile"
    if "." not in tail:
        return ""
    return _LANG_BY_EXT.get(tail.rsplit(".", 1)[-1], "")


def _foreign_language(item: dict, target: str) -> str:
    """The item's language when it DIFFERS from the target file's, else "".

    Returns "" whenever either side is unknown: we can only claim a mismatch we
    can actually establish, and a wrong "· python" tag is worse than no tag.
    Same-language hits stay unlabelled so the common case keeps a clean line.
    """
    if not target:
        return ""
    theirs = _canonical_language(item.get("language") or "")
    if not theirs or theirs == target:
        return ""
    return theirs


def _concept_doc(code: str) -> str:
    """The first doc-ish prose in `code`: docstring, else JSDoc, else leading comments."""
    m = _CONCEPT_PY_DOC.search(code)
    if m:
        return _collapse(m.group(2))

    m = _CONCEPT_JSDOC.search(code)
    if m:
        return _collapse(_CONCEPT_JSDOC_STAR.sub("", m.group(1)))

    # Only a comment block at the very TOP counts. A comment further down is
    # usually about one line of the implementation, not about the whole thing.
    m = _CONCEPT_LEADING_COMMENT.match(_CONCEPT_SHEBANG.sub("", code))
    if m:
        return _collapse(_CONCEPT_COMMENT_MARKER.sub("", m.group(0)))

    return ""


def _collapse(text: str) -> str:
    """One line, single-spaced, length-capped — embedder input, not display text."""
    return " ".join((text or "").split())[:_CONCEPT_MAX_DOC_CHARS].strip()


def concept_query(code: str) -> str:
    """Rewrite a write payload as a CONCEPT query, or "" to keep the raw payload.

    Returns something shaped like a snippet's own title — "name(params) — what it
    does" — because that is the form that measured best against the prose-forward
    snippet documents (#2242; see the table at _CONCEPT_DECL_PATTERNS).

    Returns "" rather than raising or guessing whenever there's nothing worth
    sending: no declarations and no doc, or a result too thin to beat the code it
    would replace. The caller treats "" as "use the payload as-is", so every
    unhandled language degrades to exactly the previous behaviour.
    """
    if not code or not code.strip():
        return ""

    decls: list[str] = []
    for pattern in _CONCEPT_DECL_PATTERNS:
        for match in pattern.finditer(code):
            name, params = match.group(1), match.group(2) or ""
            label = f"{name}{params}".strip()
            if label and label not in decls:
                decls.append(label)
            if len(decls) >= _CONCEPT_MAX_DECLS:
                break
        if len(decls) >= _CONCEPT_MAX_DECLS:
            break

    doc = _concept_doc(code)

    # NO DOC, NO REWRITE. An identifier alone is not a concept, and it measured
    # WORSE than the code it would replace: `collapse_into_clusters(edges)` scored
    # 0.671 against #2222 where the full code body scored 0.743. Separation from
    # the noise floor is identical (0.113 either way), but the absolute value
    # drops below the 0.68 bar — so preferring a bare name would convert a
    # comfortable hit into a miss. Undocumented code keeps the raw payload.
    if not doc:
        return ""

    head = ", ".join(decls)
    query = f"{head} — {doc}" if head else doc

    # Guard against a doc so terse it says nothing ("# TODO", "/** x */").
    if len("".join(query.split())) < _CONCEPT_MIN_CHARS:
        return ""
    return query


async def get_writepath_config(user_id: int) -> dict:
    """Write-path trigger settings: its own `enabled` and `threshold`, auto-inject's top_k.

    The threshold OVERRIDES the inherited auto-inject value — code embeddings
    have a much higher similarity floor than prose, so the two surfaces need
    different bars. See WRITEPATH_DEFAULT_THRESHOLD for the measurements (#2223).
    top_k is still shared: "how many titles at once" means the same thing on
    both surfaces, and nothing suggests they want different ceilings.
    """
    cfg = await get_autoinject_config(user_id)
    enabled_raw = await get_setting(
        user_id, WRITEPATH_ENABLED_KEY,
        "true" if WRITEPATH_DEFAULT_ENABLED else "false",
    )

    try:
        threshold = float(await get_setting(
            user_id, WRITEPATH_THRESHOLD_KEY, str(WRITEPATH_DEFAULT_THRESHOLD)))
    except (TypeError, ValueError):
        threshold = WRITEPATH_DEFAULT_THRESHOLD
    threshold = min(1.0, max(0.0, threshold))

    return {
        **cfg,
        "enabled": enabled_raw.strip().lower() in ("true", "1", "yes", "on"),
        "threshold": threshold,
    }


async def build_write_path_hint(
    user_id: int,
    path: str,
    code: str = "",
    project_id: int = 0,
    exclude_ids: list[int] | None = None,
    exclude_sync_ids: list[int] | None = None,
    stamp_shapes: list[tuple[str, str]] | None = None,
    repo_key: str = "",
    exclude_derive: list[str] | None = None,
) -> dict:
    """Prior-art hint for the plugin's PreToolUse hook on Write/Edit.

    `path` is the file about to be written, REPO-RELATIVE — matching the
    convention snippet locations are recorded in. `code` is what's about to be
    written, used only as the semantic query.

    A hit recorded AT this exact path is not a reuse suggestion — it IS the
    record of the file being changed, so it renders as the SYNC class (#2708):
    "this snippet records the file you're editing; if the edit changes the
    recorded shape, updating the record is part of the edit." That is the
    operator's chosen alternative to server-side drift flagging (decision
    #2707): the record gets corrected in the session that has the context,
    at the moment of change. Nearby and semantic hits stay the REUSE menu.

    The two classes dedup on SEPARATE channels — `exclude_ids` (reuse) and
    `exclude_sync_ids` (sync) — because they answer different questions: a
    title shown as "consider reusing this" twenty turns ago must not silence
    "you are editing the recorded file right now" (#2708).

    Carries auto-inject's anti-bloat gates (margin, session dedup,
    titles-never-bodies) plus the shared top-k cap across ALL arms — so a file
    with a lot of recorded history can't turn one edit into a wall of text. Two
    gates are its OWN, because code is not prose: a stricter similarity
    threshold, and a minimum-substance floor on `code` below which the
    semantic arm doesn't run at all (#2223 — see WRITEPATH_DEFAULT_THRESHOLD and
    WRITEPATH_MIN_CODE_CHARS). Returns empty context when disabled, when there's
    no path, or when nothing is recorded — which is the common case, and the point.

    Note the repo↔project mapping is deliberately one-way: the hook sends a git
    remote, which the ROUTE resolves to `project_id` through the repo bindings.
    It is never used as the location `repo` filter — a snippet's `repo` is a
    free-text label the operator typed ("Scribe"), not a remote URL, and matching
    one against the other would silently return nothing.

    Returns {"context": str, "note_ids": list[int], "sync_note_ids": list[int],
    "config": dict} — `sync_note_ids` is the subset of `note_ids` shown as the
    sync class, so the hook can feed each dedup channel its own ids. The
    semantic arm is logged to retrieval_logs as source='write_path' — its own
    source, so its precision is tunable separately from auto-inject's.

    Location hits still carry no score and so stay out of retrieval_logs, whose
    score distribution they would corrupt. What closed the gap (#2085) is that
    un-scored surfacing now has its own home: every arm emits note_usage_events,
    tagged 'write_path_sync' vs 'write_path_place' vs 'write_path_semantic', so
    each claim's pull-through rate is measurable on its own.

    ``stamp_shapes`` turns the same request into the ledger's write-path feed
    (#2791): the (kind, name) definitions the hook saw in — or enclosing —
    the payload. When the session has PULLED a snippet recently and this
    payload references or resembles it, those shapes land as `instance` rows
    (classified_by=hook, see shape_ledger.stamp_write_path_instances) and
    the result's ``stamped`` lists them. The route passes it only for a
    caller allowed to write — a read-scoped key gets the hint, never the
    stamp. ``repo_key`` (the hook's remote, normalised) homes a provisional
    row for a shape the ledger has not synced yet.
    """
    cfg = await get_writepath_config(user_id)
    empty = {"context": "", "note_ids": [], "sync_note_ids": [], "config": cfg,
             "stamped": [], "divergence": [], "derive": [], "derive_keys": []}
    path = (path or "").strip()
    if not cfg["enabled"] or not path:
        return empty

    top_k = cfg["top_k"]
    excluded = set(exclude_ids or [])
    sync_excluded = set(exclude_sync_ids or [])
    scope_project = project_id or None

    # --- the sync class, and arm 1 by place ---
    # `path` matches exact-or-under (see knowledge.py), and nothing sits under
    # a FILE path — so the file query returns precisely the snippets recorded
    # AT this path: the sync class. The directory query is the reuse-shaped
    # "nearby" arm, unchanged.
    here: list[dict] = []
    nearby: list[dict] = []
    try:
        here, _ = await snippets_svc.list_snippets(
            user_id, path=path, limit=top_k, project_id=scope_project,
        )
        directory = path.rsplit("/", 1)[0] if "/" in path else ""
        if directory and len(here) < top_k:
            nearby, _ = await snippets_svc.list_snippets(
                user_id, path=directory, limit=top_k, project_id=scope_project,
            )
    except Exception:
        logger.warning("Write-path location lookup failed", exc_info=True)

    # Sync hits dedup ONLY against their own channel — reuse-`excluded` ids
    # stay eligible here, which is the whole point of the split. Either way
    # they join `seen`, so the reuse arms (where the directory query would
    # surface them again) never re-list a record the sync block owns.
    seen: set[int] = set(excluded)
    synced: list[dict] = []
    for item in here:
        nid = int(item["id"])
        seen.add(nid)
        if nid not in sync_excluded and len(synced) < top_k:
            synced.append(item)

    placed: list[tuple[str, dict]] = []
    for item in nearby:
        nid = int(item["id"])
        if nid in seen:
            continue
        seen.add(nid)
        placed.append(("nearby", item))

    # The stamping feed's "actually pulled it" half (#2791). Read once, before
    # the semantic arm, because the arm's query doubles as the resemblance
    # test: a pulled snippet this session already saw (so it sits in `seen`)
    # must still be SCORED for this payload — it just isn't re-listed.
    pulled: dict = {}
    if stamp_shapes:
        pulled = await shape_ledger_svc.recent_pulls(user_id)
    resembles: dict[int, float] = {}

    # --- arm 2: by meaning ---
    scored: list[tuple[str, dict]] = []
    remaining = top_k - len(synced) - len(placed)
    query = (code or "").strip()
    # Drop payloads too small to carry meaning before spending an embedding on
    # them — a one-line Edit is not a helper being rewritten, and its embedding
    # scores off the corpus floor rather than off any real resemblance (#2223).
    # Whitespace doesn't count: code is indentation-heavy, so raw length would
    # let a deeply-nested one-liner through on padding alone.
    if len("".join(query.split())) < WRITEPATH_MIN_CODE_CHARS:
        query = ""
    # ORDER MATTERS: the floor above judges the RAW payload, this rewrites it.
    # Snippet documents are prose-forward, so a concept query out-scores the code
    # itself by a wide margin (#2242). The rewritten query is allowed to be
    # short — "slugify(t) — turn text into a url slug" is a fine query at 38
    # chars, and it only exists because the raw payload already cleared the
    # floor. Applying the floor after this would throw away the best queries.
    if query:
        query = concept_query(query) or query
    if remaining > 0 and query:
        t0 = time.perf_counter()
        # Pulled-and-seen ids stay in the query (as evidence) but never in
        # the menu — the dedup contract holds, the resemblance still lands.
        pulled_seen = seen & set(pulled)
        hits = await semantic_search_notes(
            user_id, query,
            limit=remaining + len(pulled_seen),
            threshold=cfg["threshold"],
            project_id=scope_project,
            exclude_ids=seen - pulled_seen,
            # Snippets AND recorded experience (#2246). This arm was
            # snippets-only, which is auto-inject's mistake inverted: an issue
            # saying "we tried this and it deadlocked", or a dev-log recording
            # how a problem was solved, is prior art for the code about to be
            # written — arguably better prior art than a resembling helper,
            # because it says what NOT to do.
            #
            # `task_kind="issue"` keeps the open to-do list out. A task titled
            # "add debouncing to the search box" resembles the code being
            # written and answers nothing; an ISSUE is corrective work with a
            # root cause in it, and a non-task note is durable knowledge. Both
            # earned their place; a todo did not.
            note_type=("snippet", "note"),
            task_kind="issue",
            # Same reasoning as auto-inject: nobody asked for this, so it takes
            # the browse scope and never surfaces a one-to-one direct share.
            scope="browse",
        )
        resembles = {
            int(note.id): float(score) for score, note in hits
            if int(note.id) in pulled
        }
        hits = [(s, n) for s, n in hits if int(n.id) not in seen][:remaining]
        record_retrieval(
            user_id=user_id, source="write_path", query=query,
            threshold=cfg["threshold"], limit=remaining,
            # is_task is None, not False: this arm now returns issues too, and
            # recording it as a notes-only retrieval would misdescribe the
            # candidate set the threshold is being tuned against.
            project_id=scope_project, is_task=None, results=hits,
            duration_ms=(time.perf_counter() - t0) * 1000.0,
        )
        if hits:
            top_score = hits[0][0]
            for score, note in hits:
                if score < top_score - _AUTOINJECT_BAND:
                    continue
                # Name the kind unless it's a snippet — the menu's default and
                # the header's default reading. An issue or a dev-log offered
                # here is a different KIND of claim ("this was already tried")
                # and an unlabelled line would be read as "here is code to
                # reuse", which is the opposite of what it says.
                kind = _record_kind(note)
                scored.append((
                    f"similar {score:.2f}" if kind == "snippet"
                    else f"similar {score:.2f} · {kind}",
                    {
                        "id": int(note.id), "title": note.title, "user_id": note.user_id,
                        # Carried so the line can disclose a cross-language hit
                        # (#2244). The semantic arm is where these actually arise —
                        # a snippet recorded at the path you're editing is almost
                        # never in another language, but a concept match easily is.
                        "language": (note.data or {}).get("language") if note.data else None,
                    },
                ))

    menu = (placed + scored)[:max(0, top_k - len(synced))]

    # The stamp runs whether or not anything is rendered — after dedup, the
    # common case is a silent hint and a pulled canon being instantiated.
    stamped: list[dict] = []
    if stamp_shapes and pulled:
        try:
            stamped = await shape_ledger_svc.stamp_write_path_instances(
                user_id, project_id, path=path, shapes=stamp_shapes,
                code=code or "", pulled=pulled, resembles=resembles,
                repo_key=repo_key,
            )
        except Exception:
            logger.warning("Write-path ledger stamping failed", exc_info=True)
    # The in-band button-B check (#2793): the hook named the shapes being
    # written; if this directory+kind is canon-dense and a named shape isn't
    # (about to be) an instance of that canon, say so NOW — at the write,
    # not at the next audit.
    divergence: list[dict] = []
    if stamp_shapes and project_id:
        try:
            divergence = await shape_ledger_svc.write_time_divergence(
                project_id, path, stamp_shapes, stamped
            )
        except Exception:
            logger.warning("write-time divergence check failed", exc_info=True)
    # The in-band DERIVE check (#2900): the ledger's own knowledge of the
    # names being written — a duplicate family with no canon, or a canon
    # recorded elsewhere. This is the arm the by-name local grep could not
    # be: it knows whether the other copies are canon or stray. Keyed per
    # session (`exclude_derive`) so a family is named once, not per edit.
    derive: list[dict] = []
    if stamp_shapes and project_id:
        try:
            found = await shape_ledger_svc.write_time_derive(project_id, path, stamp_shapes)
            skip = set(exclude_derive or [])
            derive = [d for d in found if d.get("key") not in skip]
        except Exception:
            logger.warning("write-time derive check failed", exc_info=True)
    if not synced and not menu and not stamped and not divergence and not derive:
        return empty

    owners = await owner_names_for({
        int(it["user_id"]) for it in synced + [it for _m, it in menu]
        if it.get("user_id") is not None and int(it["user_id"]) != user_id
    })

    def _owner_of(item: dict) -> str | None:
        owner_id = item.get("user_id")
        if owner_id is None or int(owner_id) == user_id:
            return None
        return owners.get(int(owner_id)) or "another user"

    target_lang = _language_for_path(path)
    rendered: list[tuple[dict, str, str | None, str]] = []
    for marker, item in menu:
        rendered.append((item, marker, _owner_of(item), _foreign_language(item, target_lang)))

    lines: list[str] = []
    sync_note_ids: list[int] = []
    if synced:
        # The sync framing (#2708). Deliberately imperative about the record —
        # the reuse claim ("start from this shape") is still implied by the
        # title being right there, but the load-bearing sentence is the one no
        # other surface says: keeping the record true is part of THIS edit.
        lines.append(
            f"> Recorded in Scribe AT `{path}` — the snippet(s) below record "
            "the file this edit is changing. Reuse/extend the recorded shape "
            "rather than writing a parallel one; and if this edit changes what "
            "a record captures, updating it is part of the edit: "
            "`update_snippet(id, code=…)` with the new shape, or "
            "`verify_snippet(id, status=\"ok\", commit_sha=…)` after confirming "
            "it still holds. Open with `get_snippet(id)` (shown once per session):"
        )
        for item in synced:
            sync_note_ids.append(int(item["id"]))
            lines.append(_prior_art_line(item, "records this file", _owner_of(item)))

    if menu:
        lines.append(
            f"> Prior art already recorded in Scribe for `{path}` — open one with "
            "`get_snippet(id)` for a snippet, `get_task(id)` for an issue, "
            "`get_note(id)` otherwise. Reuse a snippet rather than writing a fresh "
            "one-off; read an issue before repeating what it records "
            "(titles only; shown once per session):"
        )
    # Say what a language tag MEANS, and only when one is actually on the menu.
    # Without this the reader has to infer why "· python" is attached to a hit on
    # a .ts file, and the two ways of guessing wrong are both bad: dismiss it as
    # irrelevant, or paste Python into TypeScript. Retrieval matches on concept,
    # so these are genuinely useful — as the SHAPE of a solution, not as code.
    if any(lang for _i, _m, _o, lang in rendered):
        lines.append(
            "> A tagged language means that snippet is in a DIFFERENT language "
            "than this file — it matched on what it does, so treat it as the "
            "shape of a solution to adapt, not code to copy."
        )

    note_ids: list[int] = list(sync_note_ids)
    for item, marker, owner, foreign_lang in rendered:
        note_ids.append(int(item["id"]))
        lines.append(_prior_art_line(item, marker, owner, foreign_lang))

    if stamped:
        lines.append(_stamp_line(path, stamped))
    if divergence:
        lines.append(_divergence_line(path, divergence))
    if derive:
        lines.append(_derive_line(path, derive))

    # Split by arm, which is the whole reason this table exists. The place arm
    # carries no score and so has no home in retrieval_logs; before #2085 a
    # snippet surfaced BY PLACE left no trace anywhere, making the arm that
    # fires on the strongest possible claim ("there is already a canonical
    # helper in this exact file") the one arm nobody could measure. The sync
    # class gets its own tag: its pull-through rate is the number that says
    # whether edit-time record-sync actually happens (#2708's success measure).
    by_arm: dict[str, list[int]] = {}
    if sync_note_ids:
        by_arm["write_path_sync"] = list(sync_note_ids)
    for marker, item in menu:
        arm = "write_path_place" if marker == "nearby" else "write_path_semantic"
        by_arm.setdefault(arm, []).append(int(item["id"]))
    for arm, ids in by_arm.items():
        record_surfaced(user_id=user_id, note_ids=ids, source=arm)

    return {
        "context": "\n".join(lines),
        "note_ids": note_ids,
        "sync_note_ids": sync_note_ids,
        "config": cfg,
        "stamped": stamped,
        "divergence": divergence,
        "derive": derive,
        "derive_keys": [d["key"] for d in derive],
    }


def _derive_line(path: str, derive: list[dict]) -> str:
    """The ledger's word on the names being written (#2900): a duplicate
    family to derive, or a canon to reuse — said at the write."""
    parts = []
    for d in derive:
        if d.get("canon"):
            c = d["canon"]
            parts.append(
                f"`{c['label']}` is canon — snippet #{c['snippet_id']} at `{c['path']}`; "
                "pull it and reuse, don't redefine"
            )
            continue
        f = d["family"]
        files = ", ".join(f"`{x}`" for x in f.get("files") or [])
        more = f.get("file_count", 0) - len(f.get("files") or [])
        if more > 0:
            files += f" +{more} more"
        n = f.get("file_count", 0)
        if f.get("identical"):
            what = f"is a duplicate family with no canon — identical body in {n} other file(s)"
        else:
            # A name family: the same definition name living in several
            # files. CSS is only ever grouped this way (note 2917) — a class
            # is a recipe, and the recipe is what gets derived or dismissed.
            what = f"is a repeated name with no canon — defined in {n} other file(s)"
        # The dismissal reason the family most likely earns: a class name
        # reused for different purposes is scoped styling; a code name reused
        # across modules is convention plumbing.
        dismiss = "scoped-css" if d.get("kind") == "css" else "convention-plumbing"
        parts.append(
            f"`{f['label']}` {what}: {files}; derive it now: "
            "record the canon (create_snippet) and make the copies instances "
            "(classify_shapes) — or, if these are convention not copies, "
            f"`classify_shapes(..., status=\"exempt\", reason_code=\"{dismiss}\")` "
            "dismisses the family — rather than adding another copy"
        )
    return f"> Shape ledger at `{path}`: " + "; ".join(parts) + "."


def _divergence_line(path: str, divergence: list[dict]) -> str:
    """Button B where button A is canon — named at the write (#2793)."""
    parts = [
        f"`{('.' if d['kind'] == 'css' else '') + d['symbol']}` → #{d['canon_snippet_id']} "
        f"({d['instances']} of {d['judged']} judged siblings are its instances)"
        for d in divergence
    ]
    return (
        f"> Divergence check at `{path}`: a canon dominates this directory — "
        f"{'; '.join(parts)}. If this is a new instance, pull that snippet "
        "and build from it; if it is a deliberate departure, "
        "`classify_shapes(..., status=\"variant\", reason=…)` records the why; "
        "otherwise it reads as unintended divergence."
    )


def _stamp_line(path: str, stamped: list[dict]) -> str:
    """One line saying what the ledger just recorded, so the session can
    correct a wrong stamp in the moment rather than an audit finding it."""
    by_snippet: dict[int, list[str]] = {}
    for row in stamped:
        label = f".{row['symbol']}" if row["kind"] == "css" else row["symbol"]
        by_snippet.setdefault(int(row["snippet_id"]), []).append(f"`{label}`")
    parts = [
        f"{', '.join(names)} → instance of #{sid}"
        for sid, names in by_snippet.items()
    ]
    return (
        f"> Shape accounting: recorded at `{path}` — {'; '.join(parts)} "
        "(classified_by=hook: you pulled that snippet this session and this "
        "code references/resembles it). Not an instance? `classify_shapes` "
        "overrides a hook stamp."
    )


async def _topic_titles(topic_ids: set[int]) -> dict[int, str]:
    """Map topic_id -> title for the given ids (live topics only)."""
    if not topic_ids:
        return {}
    async with async_session() as session:
        rows = await session.execute(
            select(RulebookTopic.id, RulebookTopic.title).where(
                RulebookTopic.id.in_(topic_ids),
                RulebookTopic.deleted_at.is_(None),
            )
        )
        return {tid: title for tid, title in rows.all()}


async def build_session_context(
    user_id: int, project_id: int = 0, unbound_repo: str = ""
) -> dict:
    """Render the SessionStart context for a user, optionally project-scoped.

    Args:
        user_id: the operator.
        project_id: the resolved active project (0 = none). The endpoint
            resolves this from the working repo's remote, not from config.
        unbound_repo: when the hook sent a repo remote that maps to no project,
            its normalized key — triggers a one-line "bind this repo" hint so
            the binding is self-healing.

    Returns {"context": str, "rule_count": int, "project": dict | None}.
    `context` is markdown ready to drop into `additionalContext`; it is capped
    at _MAX_CHARS with an explicit truncation note so the hook can pass it
    through verbatim.
    """
    # Inside a project, the always-on set is the project's: an inception
    # exclusion (milestone 297) takes a rulebook out of this block, and is
    # named below so the departure is visible rather than silent.
    rules = await rulebooks_svc.list_always_on_rules(user_id, project_id=project_id)
    excluded = (
        await rulebooks_svc.excluded_always_on_rulebooks(user_id, project_id)
        if project_id else []
    )
    topic_map = await _topic_titles({r.topic_id for r in rules if r.topic_id})

    lines: list[str] = [
        "# Scribe — standing session context (auto-injected by the Scribe plugin)",
        "",
        "You are working with Scribe, the operator's self-hosted second brain. "
        "The always-on rules below are BINDING this session. Titles only — full "
        "text via `list_always_on_rules()` or `get_rule(id)`.",
        "",
        "## Always-on rules (by topic)",
    ]

    # rules already arrive ordered by rulebook/topic/order, so grouping by
    # consecutive topic_id preserves the intended sequence.
    current_topic: int | None = object()  # sentinel distinct from any id/None
    for r in rules:
        if r.topic_id != current_topic:
            current_topic = r.topic_id
            heading = topic_map.get(r.topic_id, "ungrouped") if r.topic_id else "ungrouped"
            lines.append(f"### {heading}")
        lines.append(f"- [{r.id}] {r.title}")
    if excluded:
        names = ", ".join(f"{e['title']} (#{e['id']})" for e in excluded)
        lines += [
            "",
            f"Excluded for this project by its inception decision (not binding here): {names}.",
        ]

    project_dict: dict | None = None
    if project_id:
        project = await projects_svc.get_project(user_id, project_id)
        if project is not None:
            _, open_count = await notes_svc.list_notes(
                user_id, is_task=True, status="todo", project_id=project_id, limit=1,
            )
            goal = (getattr(project, "goal", "") or "").strip()
            project_dict = {"id": project.id, "title": project.title}
            lines += [
                "",
                f"## Active project: {project.title} (id {project.id})",
                f"Goal: {goal[:200]}" if goal else "",
                f"Open todo tasks: {open_count}",
            ]

            # A design system binds the same way a rule does, and until this
            # existed it had no push channel — the standards were reachable only
            # by an agent that already knew to look for them. Summary only: the
            # token VALUES are a tool call away, and pasting a hundred of them
            # into every session would crowd out the context they inform.
            if project.design_system_id:
                design = await design_systems_svc.design_context(
                    user_id, project.design_system_id,
                )
                if design:
                    inherits = (
                        " (inherits " + " › ".join(design["inherits_from"]) + ")"
                        if design["inherits_from"] else ""
                    )
                    groups = ", ".join(design["token_groups"])
                    lines += [
                        "",
                        f"## Design system: {design['title']} "
                        f"(id {design['id']}){inherits}",
                        f"{design['token_count']} tokens"
                        + (f" across {groups}" if groups else "")
                        + ". This project's UI is built from these, not from "
                        "literals — reach for a token before writing a colour, "
                        "size, radius or duration by hand.",
                        f"Values: `resolve_design_system({design['id']})` · "
                        f"stylesheet: `get_design_system_stylesheet({design['id']})` "
                        f"· the prose (aesthetic, voice, where the accent may "
                        f"appear): `enter_project` returns it, or "
                        f"`get_design_system({design['id']})`.",
                    ]
    elif unbound_repo:
        lines += [
            "",
            "## Repository not yet bound",
            f"This repo (`{unbound_repo}`) isn't mapped to a Scribe project, so "
            "no project context was loaded. Bind it once with "
            f'`bind_repo(repo_url="{unbound_repo}", project_id=<id>)` '
            "(call `list_projects` to find the id) and future sessions here will "
            "auto-load that project's context.",
        ]

    lines += [
        "",
        "Reflex: search Scribe (search / list_tasks / list_notes, scoped to the "
        "active project) before answering or starting work; prefer UPDATING an "
        "existing note/rule over creating a new one.",
    ]

    context = "\n".join(line for line in lines if line is not None)
    if len(context) > _MAX_CHARS:
        context = context[:_MAX_CHARS].rstrip() + "\n\n…(truncated — call list_always_on_rules())"

    return {"context": context, "rule_count": len(rules), "project": project_dict}
