FabledScribe/src/scribe/services/plugin_context.py

"""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)` / `search(content_type="rule")` 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 textwrap
import time


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 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, semantic_search_rules
from scribe.services.note_usage import record_surfaced
from scribe.services.rule_usage import record_rule_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

# Max chars of the project goal on the session-start Goal line. The full goal is
# one enter_project away; a cut says so rather than ending mid-word (#4036).
_GOAL_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

# The standing-rule arm (milestone 307) gets its own bar — the split #2223 made
# one surface down, now made for the THIRD corpus. It inherited 0.68 above, and
# that number was measured against code-vs-note-PROSE. It was never re-derived
# for code-vs-RULE-TEXT.
#
# THE STRUCTURAL ARGUMENT, which is the only kind admissible here (rule 115).
# Two facts hold on any install, including one with six rules and no telemetry:
#
#   1. The eligible corpus is SMALL — every rule an install owns, still only
#      a few dozen documents against thousands of notes. A top-k over a small
#      pool always returns something, so "the best match cleared the bar"
#      drifts from "a good match exists" toward "N things were ranked". A bar
#      calibrated for best-of-thousands is cleared by best-of-forty as
#      arithmetic rather than relevance.
#      This argument WEAKENED when the arms stopped filtering to one tier
#      (see the note on that below): a larger pool makes clearing the bar
#      mean more, not less. The threshold was deliberately left where it was
#      anyway — moving two variables at once would make the resulting
#      distribution unreadable, and this one errs toward silence on purpose.
#   2. Rules are short imperative technical English — a far more HOMOGENEOUS
#      corpus than note prose. #2223 measured the floor for code against prose
#      at 0.55-0.63 and set 0.68 above it. A more homogeneous corpus has a
#      HIGHER floor, so 0.68 is not merely inherited, it is below where this
#      corpus's noise sits.
#
# WHY 0.72 AND NOT A NUMBER OFF A HISTOGRAM. The exact offset between prose's
# floor and rule-text's is not derivable in general — it depends on how an
# install writes its rules — so the default errs deliberately toward SILENCE
# rather than toward recall, on an asymmetry that is itself structural: this
# hint fires on EVERY write. A missed rule is recoverable, because the rule is
# still in Scribe and the agent can search it. A hint that cries wolf is not:
# it teaches the reader to skip the whole block, and the surface is lost along
# with the true positives it would have carried. The arm's own comment already
# says "noise on a hint that fires on every write is how a hint gets ignored".
#
# TUNE IT FROM YOUR OWN INSTANCE, which is now possible: `retrieval_telemetry`
# reports `rule_usage.pull_through` (milestone 333 step 3). Raise this if rules
# arrive unread; lower it if rules you needed never arrived. What would RETIRE
# it: a cross-encoder rerank (#1038), which would make a similarity bar the
# wrong control entirely.
# SCOPED TO THE WRITE-PATH ARM SINCE #3853. The command arm has its own bar
# below, and the measurement that separated them is recorded there. Everything
# above still holds for THIS arm: a code payload is long and rich, which is the
# case 0.72 was calibrated on, and the telemetry says it is working — the
# write-path rule arm speaks on 37% of its calls and its refused mass sits at
# p50 0.6989, comfortably under the bar rather than piled against it.
RULEHINT_THRESHOLD_KEY = "kb_rulehint_threshold"
RULEHINT_DEFAULT_THRESHOLD = 0.72

# THE COMMAND ARM'S OWN BAR, AND WHY IT IS NOT THE WRITE PATH'S (#3853).
#
# One bar served both act arms until this. They are not the same problem: a
# write-path query is a code payload, long and rich, while a pre-tool query is
# a shell command — often under a dozen words. Less text, less signal, lower
# scores for the same relevance. At a shared 0.72 the two arms measured like
# different subsystems:
#
#   write_path_rule   2,325 calls, speaks on 37%, near-miss p50 0.6989
#   pre_tool_rule    11,768 calls, speaks on  2%, near-miss p50 0.6794
#
# The second is not a quiet surface, it is a mute one: 11,530 of 11,768 calls
# said nothing, with near-miss p90 at 0.7097 — refused mass piled one
# hundredth under the line, which is the shape a bar set too high leaves. The
# note arms are the control and look nothing like it (auto_inject refuses at
# p90 0.5463, write_path at 0.6738, both far below their bars).
#
# WHAT 0.68 IS MEASURED AGAINST. Eight replayed queries, consequential acts
# against innocuous ones, scored on the post-#3855 corpus:
#
#   0.7571  git push origin dev                 consequential
#   0.7245  cd ...; git fetch; git add -A       consequential
#   0.7193  git pull --rebase origin dev        consequential
#   0.6850  docker compose up -d                consequential
#   ----------------------------------------  0.68
#   0.6735  wc -l src/*.py && date              innocuous
#   0.6544  grep -rn useState src/              innocuous
#   0.6099  sed -n '120,160p' package.json      innocuous
#   0.6056  ls -la && cat README.md             innocuous
#
# At 0.72 three of the four consequential acts retrieved NOTHING, including
# `git pull --rebase origin dev`, where rules 153, 1 and 2 all ranked
# correctly and all sat between 0.7126 and 0.7193.
#
# THE SEPARATION IS 0.0115 WIDE, and that is a caveat, not a result. Eight
# probes set a direction; they do not settle a number. `near_miss_samples` on
# a few days of post-#3855 traffic is what settles it, and this is the bar to
# re-read first.
#
# This also CORRECTS an assumption stated above. That comment argued 0.68 was
# "below where this corpus's noise sits", inferring a higher floor from the
# corpus being homogeneous. Measured, the command arm's noise ceiling is
# 0.6735 — so 0.68 clears it, barely, rather than sitting under it. The
# inference was reasonable and the measurement disagrees.
#
# WHY LOWERING IS SAFER NOW THAN IT WOULD HAVE BEEN. Until #3851 this arm had
# a single slot, so its one line had to be right and a high bar was the only
# control. The band now does noise control downstream: a marginal hit that
# clears the bar still has to score within `_RULEHINT_BAND` of the top to be
# rendered. The bar's job shrank, so the bar can.
#
# The noise floor above is set by CROSS-PROJECT BLEED rather than bad ranking
# — 0.6735 is another project's shell-command rule matching a shell command in
# this one, which is a correct match to a rule that should never have been
# eligible. Retrieval is ownership-scoped, not project-scoped. Scoping it
# would drop that ceiling and widen the 0.0115, which is the larger fix and
# the reason to settle project scoping before tuning this number twice.
TOOLRULE_THRESHOLD_KEY = "kb_toolrule_threshold"
TOOLRULE_DEFAULT_THRESHOLD = 0.68

# A SET OF RULES PER ACT, NOT THE SINGLE BEST ONE (#3851).
#
# This was 1, and the reasoning for that is kept below rather than deleted
# because it was correct for the world it was written in and the half of it
# that still holds is what shapes the replacement.
#
# THE OLD ARGUMENT. "With a corpus this small, top-k does as much damage as
# the threshold: k=2 over a few dozen candidates means the second line is
# almost always the second-best noise, arriving with the same confident
# framing as the first." True — and note the premise. Retrieval was then a
# SUPPLEMENT to a 33-rule resident set, so the arm's job was to add one
# salient rule beside everything the session already held. One was the right
# number for an accent.
#
# WHAT CHANGED. Milestone 394 removes residency, and then this arm is not the
# accent, it is the whole delivery. Moments genuinely need several rules at
# once: `git push origin dev` is governed by 1 (`dev` is home), 2 (never
# `main` unasked), 9 (poll CI) and 140 (let each action land) simultaneously,
# and each alone permits the mistake the others catch. One slot cannot serve
# that, and "the single best" is not a coherent answer when four rules bind.
#
# WHY A CAP PLUS A BAND, RATHER THAN A BIGGER CAP. The old argument's real
# content is that a fixed k invents lines — it fills slots whether or not
# anything deserves them. A band does not: it keeps what is close to the top
# and nothing else, so a moment with one clearly-relevant rule still shows
# one, and a moment with four shows four. The corpus decides, not a constant.
# The cap survives as a ceiling on the worst case, not as the usual answer.
RULEHINT_LIMIT = 5

# MEASURED, NOT REASONED — and the reasoning it replaced was wrong (#3851).
#
# The prediction was that rules would rank SHARPLY, because `rule_document()`
# shapes them the way snippets are shaped — trigger in the embedded title and
# again above the body — and note 2485 measured snippets separating their top
# hit by 0.153 while every other kind managed 0.010–0.023.
#
# They do not. Three probes against real act queries, scored by the same
# embedding the arms use:
#
#   `git push origin dev`         top 0.757, gap to second 0.022
#   `docker compose up -d`        top 0.685, gap to second 0.016
#   a bare-owner-filter query     top 0.656, gap to second 0.020
#
# That is dev-log territory, not snippet territory: rules arrive as a
# tightly-packed block. Shaping alone did not buy separation, which is worth
# recording because the opposite was the natural inference from 2485.
#
# So the band is narrow BECAUSE the corpus is flat. At 0.10 — the notes
# menu's value — every one of the top eight on the push probe falls inside,
# including a CI-registry rule and another project's branch policy. At 0.05
# it admits roughly three ranks, which is the span where the scores are still
# saying something. 2485's Finding 3 is the standing caveat: no band value
# fixes a tie, and if rules ever rank as flat as dev-logs did this control
# stops working and the answer is a reranker (#1038), not a smaller number.
_RULEHINT_BAND = 0.05

# AND A REPEAT COMPETES ON RANK ALONE (#3750), WHICH SURVIVES THE BAND.
#
# Since #3750 a hit already on the session's exclusion ledger is RENDERED
# rather than dropped, which raises a question the old behaviour never had to
# answer: when the top-ranked hit is one the session has already seen, does it
# take its place, or step aside for a fresh rule behind it?
#
# It keeps its place, and nothing is promoted past it. #3851 widened the arm
# from one slot to a banded set and did NOT reopen this: a repeat still ranks
# where it ranks, and the band is applied to scores with no regard for what
# the session has seen. The two are independent, exactly as `kind` and `seen`
# are independent in the renderer — rank answers "what is relevant now" and
# the ledger answers "have you been told", and neither is evidence about the
# other. Two reasons, both unchanged by the widening.
#
# RANK IS THE ANSWER TO "WHAT IS RELEVANT NOW". If the repeat scores 0.85 and
# the best fresh candidate 0.73, the repeat is the better match for the action
# actually being taken. Overfetching in order to promote the fresh one past it
# would reinstate exactly the withholding this milestone exists to remove, one
# rank deeper and harder to see — recency is not a reason to show a worse
# match, and "you have seen this" is not the same claim as "you are holding
# this".
#
# AND THE COST OF A SECOND LINE IS NOW PAID DIFFERENTLY, NOT WISHED AWAY.
# This paragraph used to read "a second line is the one thing the limit above
# forbids", on the strength of "a reference costs the same ~40 tokens as a
# first surfacing". Both halves are now wrong and the second was already
# wrong when written: a full line is ~143 tokens once the trigger is rendered,
# and #3855's rewrite of the corpus roughly tripled trigger length, so five
# full lines on a push probe measure ~646 tokens before EVERY Bash call.
#
# That number is why the widening pairs with a compact rendering rather than
# arriving alone (see `_rule_hint_line`). The old paragraph's instinct — that
# a fourth voice which speaks at full volume twice is where a reader stops
# reading — is the half worth keeping, and it is answered by making the
# later lines quieter rather than by refusing to have them. Top hit full,
# the rest as references: ~299 tokens against ~568 for five full lines, so
# roughly 2x the old single line for four more rules.
#
# The reference keeps its TAIL and loses only its trigger, which is both the
# cheaper and the safer cut — see `_rule_hint_line`, where the first attempt
# dropped the tail as well and #3750's guard caught it within one commit.
#
# The consequence is deliberate and worth naming: a rule that keeps ranking
# first for a recurring situation keeps being referenced, every time the
# situation recurs. That is the intended behaviour — the situation recurring
# IS the trigger — and its decay belongs to exclusion ageing (#3751), not to
# a rule that ranks first being quietly demoted for having won before.

# WHY THE ARMS NO LONGER FILTER TO ONE TIER (#3702).
#
# Both arms used to pass `tier="conditional"`, on the reasoning that an
# always-on rule is already in the session, so surfacing it again is pure
# noise. That reasoning conflates two different things:
#
#   PRESENT IN CONTEXT  — the rule was delivered at session start.
#   SALIENT AT THE MOMENT  — the rule is in front of the reader when the
#                            action it governs is about to be taken.
#
# A rule handed over in a list at turn zero is present while a session writes
# a config value three hundred turns later. It is not surfaced. So the filter
# did not merely skip a redundant hint — it made a whole class of rules
# permanently ineligible for the only mechanism that puts a rule in front of
# an agent AT the moment, and the more important a rule is, the more likely
# it was in that class.
#
# The deeper defect is that the filter was doing the THRESHOLD's job. Whether
# a rule belongs in this hint is a relevance question, and a similarity bar is
# the control for relevance. A categorical exclusion standing in for a
# relevance judgment cannot be tuned, cannot be measured, and cannot be wrong
# in a way anybody notices.
#
# THIS IS A MEASURED CHANGE, NOT A SETTLED ONE. The old comment's fear is
# real — a hint that fires on every write and says obvious things teaches the
# reader to skip the block, and the surface is then lost along with its true
# positives. That fear had simply never been checked. `retrieval_logs` already
# records top_score, result_count and the query for every call, so the
# evidence now arrives on its own:
#
#   - rules clear the bar often and at high scores  -> the fear was justified,
#     the filter was a crude proxy for a bar set too low, and the WORK IS THE
#     BAR. Any reinstated filter should then carry a measured reason.
#   - rules clear rarely, in a thin band near the bar -> the filter was never
#     the right instrument and relevance was always sufficient.
#
# Only the eligibility moved. The bar and k=1 were both left exactly where
# they were, so the resulting distribution has one cause.

# How much of a command reaches the embedding (#3476). A shell call is not a
# file: most are short, and the ones that are not are usually a heredoc or a
# pasted script whose bulk says nothing about which rule applies. The VERB AND
# ITS TARGET sit at the front — `curl https://git.fabledsword.com/api/...`,
# `docker compose up`, `git checkout -b` — and that head is the whole signal.
# Sending the tail as well would push it out of a 512-token window and let a
# heredoc's prose decide the match.
_TOOL_QUERY_CHARS = 400

# 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

# --- the prompt-boundary rule arm (#3852) ------------------------------------
#
# Both existing rule arms are keyed on something the session is about to DO —
# a file write, a command. A rule that governs what to SAY has no such moment.
# Extract intent from loose phrasing, raise a conflict before acting, hand off
# an action with its reason, end a finding with an offer: every one binds on a
# RESPONSE, and no tool call precedes a response.
#
# The operator's message is the only query that exists before one is composed,
# and this arm is what runs against it. Until now that hook searched notes
# alone, so no rule had ever been retrieved against a thing the operator said.
PROMPTRULE_THRESHOLD_KEY = "kb_promptrule_threshold"
# INHERITED FROM THE ACT ARMS, AND NOT YET EARNED HERE. 0.72 was tuned against
# code and shell commands. An operator's prose is a different query shape
# against the same documents, and nothing yet says the two distributions line
# up — triggers are written in the vocabulary of the MOMENT, which for most
# rules is act vocabulary, so prose may well score lower across the board.
#
# Starting at the act arms' number anyway is deliberate: it is the only value
# with evidence behind it, and guessing lower would put an unmeasured bar in
# front of a corpus that binds. Every call is logged under `prompt_rule` from
# the first deploy, so a few days of real traffic settles it — read
# `near_miss_samples` (#3807) before moving this, not the percentile alone.
PROMPTRULE_DEFAULT_THRESHOLD = 0.72

# MORE THAN THE ACT ARMS' SINGLE SLOT, anchored on this hook's budget rather
# than theirs. RULEHINT_LIMIT is 1 because that arm fires before EVERY Bash
# call, where a second line is a second interruption per command. This arm
# fires once per TURN, on the same hook whose notes menu already spends
# AUTOINJECT_DEFAULT_TOP_K slots — so that is the comparable budget.
#
# And a prompt genuinely contains more than one act. "Merge to main and then
# start on X" is two, governed by different rules; k=1 cannot serve that case
# at all, where the act arms never face it because a command is one thing.
PROMPTRULE_LIMIT = 3

# THE COMPLETION-REPORT ARM'S OWN BAR (services/reply_preferences.py).
#
# It borrowed PROMPTRULE_THRESHOLD_KEY when it shipped, which made the two
# arms one dial: an operator lowering the bar for their own prose moved this
# one with it, silently. That contradicts the rule every other bar here
# follows — one number cannot serve arms whose queries are different shapes —
# and this arm's query is the most different of all. The others score an
# operator's prose or a session's code, both of which vary per call; this one
# scores a FIXED string (`COMPLETION_QUERY`) against rule triggers, so its
# score for a given corpus is a constant. A constant that lands under the bar
# is not a quiet arm, it is a dead one, and nothing about the prose arm's
# traffic would ever reveal it.
#
# Kept at the prose arm's starting value rather than tuned: the split is what
# makes the two independently movable, and a default is a product decision
# that this install's corpus cannot settle (rule 115).
REPORTPREF_THRESHOLD_KEY = "kb_reportpref_threshold"
REPORTPREF_DEFAULT_THRESHOLD = 0.72


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()
    _rep: dict = {}
    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",
        report=_rep,
    )
    # 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,
        best_available=_rep.get("best_available_score"),
        best_available_id=_rep.get("best_available_id"),
        searched=bool(_rep.get("searched", True)),
        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()
    _rep_ai: dict = {}
    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",
        report=_rep_ai,
    )
    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,
        best_available=_rep_ai.get("best_available_score"),
        best_available_id=_rep_ai.get("best_available_id"),
        searched=bool(_rep_ai.get("searched", True)),
        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}


async def _reserve_slot_for_preference(
    user_id: int,
    query: str,
    hits: list,
    *,
    threshold: float,
    project_id: int,
    already: set[int],
) -> tuple[list, int | None]:
    """Guarantee a preference one slot, if one clears the bar (#3894).

    THE ASYMMETRY THIS EXISTS FOR. A rule and a preference are not equally
    served by a shared score contest, because their losses are not equal:

      - a RULE crowded out here can still fire at the act arm. A `git push`
        reaches `pre_tool_rule`, a file write reaches `write_path_rule`. The
        prompt hit is a preview of a second chance.
      - a PREFERENCE about how to answer has no second chance. There is no
        later act — the response IS the act — so crowded out here it is never
        delivered at all.

    A straight ranking therefore favours the record whose loss is recoverable
    over the one whose loss is total, and it does so INVISIBLY: the rule that
    won is a legitimate hit, the telemetry looks healthy, and the only symptom
    is a preference that quietly never arrives. `reuse_slot` exists for the
    same shape one corpus over (#2463), where snippets kept losing to project
    records that merely resembled the query.

    THE SLOT BUYS POSITION, NOT A LOWER BAR — same as `reuse_slot`, which also
    reserves at `cfg["threshold"]`. A weak preference cannot buy the slot, so
    silence stays the default and the reserved line is never worse than the
    ones it sits beside. If `preference_slot` later shows a stream of
    near-misses, `best_available_id` (#3807) names which preference was
    refused and a separate bar becomes an argument with evidence behind it
    rather than a knob added on a guess.

    LEDGER REPEATS STILL COUNT AS REPRESENTED. A preference already on the
    session's ledger occupies the slot rather than being skipped for a fresh
    one: it is still rendered (#3750), just with the tail that says so, and a
    preference is the kind of record where being reminded is the point.

    Returns the possibly-extended hit list, and the id the slot spent — the
    caller needs that to keep each source's surfaced set matching its own log
    row (#3668), since the slot logs under its own name.
    """
    if any(rule.kind == "preference" for _s, rule in hits):
        return hits, None

    _t0 = time.perf_counter()
    _rep: dict = {}
    # KIND-FILTERED, so the query can only answer with what the slot is for.
    # Verifying the kind afterwards would be weaker: an unfiltered search that
    # happened to return a rule would spend the slot on it, and the line would
    # be indistinguishable from one that earned its place.
    found = await semantic_search_rules(
        user_id, query, limit=1, threshold=threshold,
        kind="preference", report=_rep, project_id=project_id or None,
    )
    fresh = [(s, r) for s, r in found if r.id not in already]
    # ITS OWN SOURCE, and both sides of the trade logged. #2463's own finding
    # is the warning rather than the precedent here: the hit that slot pushed
    # OUT was in retrieval_logs while the query that pushed it out was not, so
    # the slot could never be judged against what it displaced. `results` is
    # fresh-only, matching what gets recorded as surfaced below (#3752/#3668).
    record_retrieval(
        user_id=user_id, source="preference_slot", query=query,
        threshold=threshold, limit=1, project_id=project_id,
        is_task=None, results=fresh,
        best_available=_rep.get("best_available_score"),
        best_available_id=_rep.get("best_available_id"),
        searched=bool(_rep.get("searched", True)),
        suppressed=len(found) - len(fresh),
        duration_ms=(time.perf_counter() - _t0) * 1000.0,
    )
    seen = {rule.id for _s, rule in hits}
    slot = [(s, r) for s, r in found
            if r.kind == "preference" and r.id not in seen][:1]
    if not slot:
        return hits, None

    slot_id = int(slot[0][1].id)
    if slot_id not in already:
        record_rule_surfaced(
            user_id=user_id, rule_ids=[slot_id], source="preference_slot",
        )
    # IT EXTENDS, IT NEVER DISPLACES — and here it parts company with
    # `reuse_slot`, which evicts its menu's weakest hit. The reason is the
    # ledger rather than taste. A displaced hit was RETURNED by the general
    # search and is sitting in that call's `retrieval_logs` row, but would not
    # have been shown — so `prompt_rule`'s surfaced set would stop matching
    # its own log row, and #3668's identity would break for a reason nothing
    # in the data explains. That identity is the cheapest true statement
    # available about this pair of tables, and milestone #379 is what it costs
    # to lose it: five steps planned against a gap that was two counters
    # disagreeing, not a write path dropping rows.
    #
    # The price is one extra line, only when the general search already filled
    # the limit AND a preference cleared the bar without placing. Cheap, and
    # it buys a surface whose two tables can always be checked against each
    # other.
    return hits + slot, slot_id


async def build_prompt_rule_hint(
    user_id: int,
    query: str,
    *,
    project_id: int = 0,
    exclude_rule_ids: list[int] | None = None,
    held_rule_ids: list[int] | None = None,
) -> dict:
    """Rules and preferences that may apply to what the operator just asked.

    The third rule arm, and the one that closes a gap the other two cannot
    reach. `write_path_rule` is keyed on code, `pre_tool_rule` on a command —
    both are things the session is about to DO. A rule that governs what to
    SAY has no such trigger, and residency was the only surface it ever had.
    Removing residency (milestone 394) without this would drop that half of
    the corpus on the floor.

    A SEPARATE FUNCTION, not a branch inside build_autoinject_hint, and the
    reason is its early returns. That arm bails when auto-inject is disabled,
    when the query is blank, when nothing clears the note bar — and every one
    of those is a statement about NOTES. Folded in, a user who turned the
    notes menu off would silently lose their rules too, which is the kind of
    coupling nothing downstream could see. Two functions, two sets of gates,
    composed by the caller.

    THE OUTPUT IS DELIBERATELY NOT QUOTED, where the notes menu is. The task
    asked whether the two share a header; the answer is that neither needs
    one. A note line is a bare title and needs the menu's header to say what
    it is doing there, while a rule line names itself in its opening words
    ("Standing rule that may apply…" / "Preference that may apply…"). Leaving
    rules unquoted separates the two claims visually with no extra prose, and
    matches how a rule line already renders on both act arms.

    Fails open and returns empty context on any error, like its siblings: a
    recall aid may never break the operator's prompt.
    """
    out: dict = {"context": "", "rule_ids": []}
    q = (query or "").strip()
    if not q:
        return out

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

        t0 = time.perf_counter()
        _rep: dict = {}
        # SCOPED TO THIS SESSION'S PROJECT (milestone 414): global rules plus
        # the bound project's own. An unbound session (project_id 0) gets
        # global rules only. This arm used to search every rule the user owned,
        # so each project's rules were injected into every other project's
        # sessions — this surface speaks unasked, and a whole-rulebook answer
        # is only right for someone who asked the whole rulebook.
        hits = await semantic_search_rules(
            user_id, q, limit=PROMPTRULE_LIMIT, threshold=threshold,
            report=_rep, project_id=project_id or None,
        )
        duration_ms = (time.perf_counter() - t0) * 1000.0

        already = set(exclude_rule_ids or [])
        held = set(held_rule_ids or [])
        fresh = [(score, rule) for score, rule in hits if rule.id not in already]

        # BEFORE the early return, for the reason both sibling arms spell out
        # at length: a call that found nothing is the only evidence a bar is
        # too high, and an arm that logs only the calls it liked reports a
        # flawless clear-rate however badly it is tuned. This bar is inherited
        # and unverified for this corpus, so the zero rows are the point.
        record_retrieval(
            user_id=user_id, source="prompt_rule", query=q,
            threshold=threshold, limit=PROMPTRULE_LIMIT,
            project_id=project_id,
            is_task=None, results=fresh, duration_ms=duration_ms,
            best_available=_rep.get("best_available_score"),
            best_available_id=_rep.get("best_available_id"),
            searched=bool(_rep.get("searched", True)),
            suppressed=len(hits) - len(fresh),
        )
        # THE RESERVED SLOT RUNS BEFORE THE BAIL-OUT, and that ordering is
        # load-bearing rather than tidy. An empty general result is not proof
        # that no preference qualifies: the general search overfetches by
        # distance and then collapses, so a preference ranked below that
        # window is invisible to it while a kind-filtered query finds it at
        # once. Bailing first would make the slot dead in exactly the corpus
        # it exists for — one where rules outnumber preferences.
        hits, slot_id = await _reserve_slot_for_preference(
            user_id, q, hits, threshold=threshold,
            project_id=project_id, already=already,
        )

        # `hits`, not `fresh` (#3750): a call whose only hit is a repeat still
        # has something to say, it just says it differently.
        if not hits:
            return out

        lines = [
            _rule_hint_line(
                rule, where="to this request",
                seen=rule.id in already, held=rule.id in held,
            )
            for _score, rule in hits
        ]
        # FRESH-ONLY (#3752). A reference is a rendering decision, not a
        # retrieval outcome, and counting one here would inflate the
        # denominator pull_through is read from.
        #
        # `fresh` is the PRE-SLOT list on purpose: it is exactly what this
        # call's own `retrieval_logs` row recorded, so the two stay equal
        # (#3668). The slot's hit is surfaced under `preference_slot` by the
        # helper, against that source's own row.
        rule_ids = [rule.id for _score, rule in fresh]

        # RANKED, not ambient: this arm chose what it showed. The name is also
        # in `rule_usage.RANKED_SOURCES`, and it has to be — a ranked source
        # missing from that tuple is counted as a bulk delivery nobody decided
        # on, which silently moves it out of the pull-through denominator.
        if rule_ids:
            record_rule_surfaced(
                user_id=user_id, rule_ids=rule_ids, source="prompt_rule",
            )
        out["context"] = "\n".join(lines)
        out["rule_ids"] = rule_ids
    except Exception:
        logger.debug("prompt rule arm failed", exc_info=True)
    return out


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

    try:
        rule_threshold = float(await get_setting(
            user_id, RULEHINT_THRESHOLD_KEY, str(RULEHINT_DEFAULT_THRESHOLD)))
    except (TypeError, ValueError):
        rule_threshold = RULEHINT_DEFAULT_THRESHOLD
    rule_threshold = min(1.0, max(0.0, rule_threshold))

    try:
        tool_rule_threshold = float(await get_setting(
            user_id, TOOLRULE_THRESHOLD_KEY, str(TOOLRULE_DEFAULT_THRESHOLD)))
    except (TypeError, ValueError):
        tool_rule_threshold = TOOLRULE_DEFAULT_THRESHOLD
    tool_rule_threshold = min(1.0, max(0.0, tool_rule_threshold))

    return {
        **cfg,
        "enabled": enabled_raw.strip().lower() in ("true", "1", "yes", "on"),
        "threshold": threshold,
        # Its own bar, for a third corpus — see RULEHINT_DEFAULT_THRESHOLD.
        "rule_threshold": rule_threshold,
        # And the COMMAND arm's own bar again, for the same reason one level
        # down: a shell command is a different query shape from a code payload
        # and scores lower for the same relevance (#3853). Separate keys, so an
        # install can move one without the other — which is the whole finding.
        "tool_rule_threshold": tool_rule_threshold,
    }

def _rule_band(hits: list) -> list:
    """The top hit, plus every hit within `_RULEHINT_BAND` of it (#3851).

    The instrument that lets an act surface a SET without inventing one: a
    fixed k fills its slots whether or not anything deserves them, while this
    keeps only what the scores say is close, so one clearly-relevant rule
    still shows one and four competing rules show four.

    Sync and pure, and deliberately its own function rather than a comparison
    written twice — the two act arms are the pair #3497 records drifting apart
    by being modelled on each other instead of sharing.

    Takes `(score, rule)` pairs already ordered best-first, as both
    `semantic_search_*` helpers return them.
    """
    if not hits:
        return []
    top = hits[0][0]
    return [(s, r) for s, r in hits if s >= top - _RULEHINT_BAND]


def _rule_hint_line(
    rule, *, where: str, seen: bool, held: bool = False, compact: bool = False,
) -> str:
    """One rule hint line — both arms, both tails, both kinds (#3750, #3849).

    THREE INDEPENDENT AXES SINCE #3851. `compact` joins `kind` and `seen`, and
    like them it reads none of the others: it says how much ROOM this line
    gets, which is a fact about its rank among today's hits rather than about
    the rule. A compact line is still a full claim that the rule may apply —
    it simply cites the rule instead of quoting its trigger. Crucially it
    still carries the `seen` tail, so the three axes stay genuinely
    independent: shortening a line must not decide what it says about whether
    the session is holding the rule.

    WHY THE LATER LINES ARE QUIETER. Measured at #3851: a full line runs ~143
    tokens once the trigger is rendered, and #3855 tripled trigger lengths
    across the corpus, so five full lines cost ~568 tokens before every Bash
    call. Top-full-plus-references costs ~299 — about 2x the old single line,
    for four more rules. The budget argument that once justified a single slot
    was real; what it actually forbids is four voices at full volume, not four
    voices.

    The top hit keeps the full rendering because it is the one the ranker is
    most confident about, and a reader who acts on exactly one line should
    have acted on that one.

    TWO INDEPENDENT AXES. `kind` decides the head, `seen` decides the tail,
    and neither reads the other. A preference and a rule differ in force; a
    repeat and a first surfacing differ in whether the session already holds
    the line. Those are unrelated facts, and keeping them unrelated in the
    code is what stopped the second kind from reopening the repeat question.

    ONE FUNCTION BECAUSE THE TAILS MUST NOT DRIFT. The two arms phrase their
    heads differently ("may apply here" vs "may apply to this Bash call") and
    that difference is deliberate. Everything after it must not differ, and
    #3497's history is that the pre-tool arm inherited a defect from its
    sibling by being modelled on it rather than sharing with it. Two copies of
    a two-branch string is how one branch gets fixed and the other does not.

    WHY A REPEAT GETS A LINE AT ALL. Both arms used to drop a hit whose id was
    already on the session's exclusion ledger and emit nothing. That is correct
    only while the session still HOLDS what it was told, and a compaction
    breaks exactly that: the earlier injection is summarized away while the id
    stays on the ledger, so the rule is absent from context AND unreachable for
    the rest of the session (#3749 closes the compaction half; this closes the
    ordinary half, where a session simply stops holding a line it read an hour
    ago).

    Only ONE CLAUSE of the original line is false on a repeat — the claim that
    the rule is not in the session's loaded set. So only that clause changes.
    Title, trigger and pull pointer are identical either way, the statement is
    never injected either way, and a repeat therefore costs the same ~40 tokens
    as a first surfacing and no more.

    DELIBERATELY NOT ASKING THE SESSION WHETHER IT HOLDS THE RULE. A model
    asked "do you still hold rule 156?" will say yes, and the claim is
    unverifiable self-report about its own context. The answer is also not
    needed: the line is cheap enough to always emit and carries its own remedy
    in both branches. Removing the question removes the fragility rather than
    managing it.
    """
    trigger = (rule.when_to_apply or "").strip()
    preference = rule.kind == "preference"
    # KIND CHANGES THE HEAD; `seen` CHANGES THE TAIL. The two axes are
    # independent and stay that way, which is what lets the repeat logic above
    # survive a second kind without being reasoned about again: whether a
    # record is already on the ledger has nothing to do with how much force it
    # carries, so the seen-branch is shared verbatim.
    #
    # The noun is the whole of the visual difference, and that is deliberate.
    # A reader skimming an injected block gets one word to place the register \u2014
    # so it is the SECOND word that moves, and it is the word naming force.
    # Everything structural after it is identical, so the three kinds read as
    # one set rather than three formats (milestone 385's step 5 writes the
    # lesson voice against these two; they are one paragraph, not three).
    noun = "Preference" if preference else "Standing rule"
    # The only other place force is asserted. A rule's line tells the reader
    # not to dismiss it unread, because dismissing a rule unread is how the
    # thing it prevents happens. A preference makes no such claim: it says
    # where to find how this has been done, and following it is what keeps
    # things consistent rather than what keeps them correct.
    reason = (
        "for how this has been done before" if preference
        else "before deciding it does not apply"
    )
    # THREE STATES, BECAUSE TWO OF THEM WERE BEING TOLD THE SAME LIE (#4100).
    #
    # `seen` means an arm NAMED this rule earlier. It does not mean the session
    # read it — the line is a teaser, and a teaser skimmed past leaves nothing
    # behind, least of all after a compaction summarises the turn it arrived
    # in. "You saw it earlier this session" asserted something about the
    # reader's context that the server had no way to know.
    #
    # `held` is the observable half: a PostToolUse hook watches for the
    # `get_rule` call itself, so this is a recorded EVENT rather than a claim.
    # That distinction is what keeps the non-goal above intact — the objection
    # was to asking a model about its own context, not to noticing what it did.
    #
    # The middle state is the honest one and the one that was missing: named,
    # not opened. It gets the full invitation, because a session that skipped
    # the teaser is in almost the same position as one that never saw it.
    if held:
        tail = (
            f"You opened it earlier this session; pull it with "
            f"get_rule({rule.id}) again if you no longer hold it."
        )
    elif seen:
        tail = (
            f"Mentioned earlier this session but not opened — read it with "
            f"get_rule({rule.id}) {reason}."
        )
    else:
        tail = (
            f"Read it with get_rule({rule.id}) {reason}; it is not in this "
            "session's loaded set."
        )
    if compact:
        # THE TRIGGER GOES; THE TAIL STAYS. Only one of the two is expensive \u2014
        # a trigger runs 300-400 characters after #3855, the tail about 100 \u2014
        # so dropping the trigger is nearly the whole saving and dropping the
        # tail would be mostly sacrifice.
        #
        # It would also destroy the one thing #3750 exists to say. The tail is
        # what tells a reader whether they were already told this and may no
        # longer be holding it, and that is the entire difference they can act
        # on; a reference with no tail reads as a first surfacing whether it is
        # one or not. This branch shipped without it for one commit and
        # test_a_rule_the_session_already_holds_is_referenced_not_re_offered
        # caught it, which is the guard working exactly as #3750 intended.
        return (
            f"Also \u2014 {noun.lower()} \u201c{rule.title}\u201d. {tail}"
        )
    return (
        f"{noun} that may apply {where} \u2014 \u201c{rule.title}\u201d"
        + (f" ({trigger})" if trigger else "")
        + f". {tail}"
    )


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,
    exclude_rule_ids: list[int] | None = None,
    held_rule_ids: list[int] | 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": [],
             "rule_ids": []}
    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)
        _rep_wp: dict = {}
        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",
            report=_rep_wp,
        )
        resembles = {
            int(note.id): float(score) for score, note in hits
            if int(note.id) in pulled
        }
        shown = [(s, n) for s, n in hits if int(n.id) not in seen]
        # WHAT THIS ARM WITHHELD AFTER THE SEARCH ANSWERED, and the reason
        # `best_available_score` cannot always be reported here (#3739 again,
        # from the side its fix did not reach).
        #
        # This arm is the one note arm that filters TWICE. `exclude_ids` takes
        # `seen - pulled_seen` into the search, but the pulled-and-seen ids stay
        # in the query deliberately — `resembles` above needs them — and are
        # dropped in the line above instead. So the score the search reported is
        # PRE that drop while the row's `result_count` is POST it, and a record
        # the session had already been shown could be logged as something the
        # BAR turned away. Live proof on the first read after #3739 shipped:
        # write_path's near-miss max was 0.822 while the lowest score it ever
        # RETURNED was 0.6857 — a "rejection" that beat every acceptance.
        #
        # The suppression column cannot rescue it the way it does for the rule
        # arms: this arm's count would be PARTIAL, covering only the drops made
        # here and not the ones `exclude_ids` made inside the search, and a
        # partial number under a name that reads as complete is the substitution
        # this whole milestone exists to stop.
        #
        # So the honest answer is null — "not measured on this call" — whenever
        # this filter removed anything, because then the bar is not the only
        # thing that turned something away and the reported score may belong to
        # a record we withheld ourselves. Calls where nothing was dropped keep
        # reporting it, which is most of them.
        withheld_here = len(hits) - len(shown)
        hits = shown[: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,
            best_available=(
                None if withheld_here else _rep_wp.get("best_available_score")
            ),
            # Withheld on the SAME condition as the score. A surviving id
            # beside a null score would name a record without saying what it
            # scored, which is the pair disagreeing in the other direction.
            best_available_id=(
                None if withheld_here else _rep_wp.get("best_available_id")
            ),
            searched=bool(_rep_wp.get("searched", True)),
            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)
    staleness: list[str] = []
    # ── Have the rules moved under this session? (milestone 323) ───────
    #
    # THE RULES-ETAG STALENESS ARM IS GONE (milestone 394).
    #
    # It took a marker the session had been given at SessionStart, compared it
    # against the resident set as it stood now, and said which rules had moved
    # or fallen out of force. That was worth doing while a session held a
    # fixed set of rules from turn zero and could be holding a stale copy of
    # it hours later.
    #
    # Nothing is resident now. A rule is retrieved at the moment it applies,
    # so a session cannot be holding an out-of-date one — the next act that
    # needs it fetches it again. The staleness this arm reported was an
    # artifact of the delivery model rather than a fact about the corpus, and
    # it goes with the model.
    #
    # `staleness` survives as the list the arms below still append to.


    # The guard sits BELOW the staleness arm on purpose. A rules change is
    # unconditional news — it does not become less true because this
    # particular write happened to match no prior art — and this arm is one
    # indexed query, only when the session actually sent a marker.
    #
    # The standing-rule arm further down is deliberately left on the far side
    # of this guard: that one runs a SEMANTIC search, and moving it here would
    # run an embedding query on every write in the session. Its gating is a
    # separate question from this one (see the note on #3244).
    if not staleness and 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)))

    # Seeded with the staleness line, which is decided above the early
    # return and so cannot wait for this list to exist.
    lines: list[str] = list(staleness)
    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)

    # ── Standing rules that may apply here (milestone 307) ──────────────
    #
    # A SUGGESTION, not a binding surface, and the distinction is the design
    # (D7): a rule BINDS by being tagged to an area the project works in,
    # resolved deterministically at enter_project. This arm reaches for
    # something weaker and still useful — a conditional rule whose trigger
    # resembles what is being written, noticed at the moment it is relevant
    # rather than by being resident in every session.
    #
    # EVERY TIER, since #3702 — see the note at RULEHINT_LIMIT. This comment
    # used to read "CONDITIONAL ONLY: an always-on rule is already in the
    # session, so repeating it here would be noise". That conflated being
    # PRESENT in context with being SALIENT at the moment the action is taken,
    # and it is the same conflation #3750 corrects one layer up: a rule the
    # session was told about an hour ago is not a rule in front of the reader
    # now.
    #
    # Fails open like every other arm: a rule hint must never break a write.
    rule_ids: list[int] = []
    try:
        already = set(exclude_rule_ids or [])
        held = set(held_rule_ids or [])
        # Timed like the notes arm above. Without this the rule row was the one
        # source in the whole readout reporting a null p90_duration_ms (#3311)
        # — a gap that reads as "this surface is somehow not measurable" rather
        # than "nobody passed the number".
        rule_t0 = time.perf_counter()
        _rep_wpr: dict = {}
        hits = await semantic_search_rules(
            user_id, code or path, limit=RULEHINT_LIMIT,
            threshold=cfg["rule_threshold"],
            report=_rep_wpr, project_id=project_id or None,
        )
        rule_ms = (time.perf_counter() - rule_t0) * 1000.0
        # BAND FIRST, dedup second, and the order is the whole point (#3851).
        # The band is a statement about the SCORES — what the ranker thinks is
        # close to the best match — so letting the ledger reorder it would let
        # "you were told this already" change what counts as relevant. Those
        # are the independent axes the renderer keeps apart.
        kept = _rule_band(hits)
        fresh = [(score, rule) for score, rule in kept if rule.id not in already]
        # EVERY kept hit gets a line; `already` only changes the tail (#3750),
        # and rank only changes how much room it gets (#3851).
        for idx, (_score, rule) in enumerate(kept):
            lines.append(
                _rule_hint_line(
                    rule, where="here", seen=rule.id in already,
                    held=rule.id in held,
                    compact=idx > 0,
                )
            )
        # `rule_ids` stays FRESH-ONLY, and that is the whole telemetry story of
        # this change (#3752). It is what the hook writes to the exclusion
        # ledger and what `record_rule_surfaced` counts; a referenced rule is
        # already on the ledger by definition, and counting it as a surfacing
        # would inflate pull_through's denominator with a choice this arm never
        # made. A reference is a RENDERING decision, not a retrieval outcome.
        rule_ids.extend(rule.id for _score, rule in fresh)
        # TWO tables, and the split is not arbitrary. retrieval_logs is one
        # row per CALL, keyed on the score distribution a threshold is tuned
        # from. rule_usage_events is one row per RULE per event, which is the
        # grain "was this hint ever acted on" needs and the grain a JSONB
        # result_ids array cannot be indexed at.
        #
        # This comment used to say rule ids had nowhere to go — that
        # note_usage_events remaps ids on restore, so a rule id there would
        # return attached to whatever note took that number. That is still
        # true of the NOTE table, and it is exactly why rule_usage_events is
        # its own (milestone 333 step 1). The gap it described is closed.
        #
        # THE CALL LOG IS UNCONDITIONAL; THE SURFACING LOG IS NOT, and the
        # asymmetry is the correction #3497 exists to make. Both used to sit
        # inside an `if fresh:`, which is how this arm came to report
        # `zero_result_calls: 0` and `cleared_threshold: 133/133` — not a
        # perfectly tuned surface but one structurally unable to record its
        # own misses. #3311 read that artifact as a measurement and a whole
        # milestone was scoped on it. A call that found nothing is the ONLY
        # evidence a threshold is set too high, and it is the row every note
        # surface has always written (write_path: 421 zeroes of 613 calls;
        # auto_inject: 114 of 326). A SURFACING is different in kind: nothing
        # was shown, so no such event occurred, and its log stays guarded.
        #
        # `results=fresh`, not `hits`: the note arms pass their exclusions
        # INTO semantic_search_notes, so what they log is already
        # post-exclusion. semantic_search_rules takes no such parameter and
        # this filter is where the equivalent happens — logging `hits` would
        # quietly make this row mean something other than every other row in
        # the same readout.
        record_retrieval(
            user_id=user_id, source="write_path_rule", query=code or path,
            threshold=cfg["rule_threshold"], limit=RULEHINT_LIMIT,
            project_id=project_id,
            is_task=None, results=fresh, duration_ms=rule_ms,
            best_available=_rep_wpr.get("best_available_score"),
            best_available_id=_rep_wpr.get("best_available_id"),
            searched=bool(_rep_wpr.get("searched", True)),
            # FOUND BUT NOT SHOWN, which since #3851 has TWO causes: the
            # session had already been told (the ledger), or the score fell
            # outside `_RULEHINT_BAND` of the top hit. Both are counted here
            # because the question this answers is unchanged — a zero row must
            # be able to say whether the bar was too high or whether the arm
            # simply chose not to speak, and only the first is a reason to
            # move the threshold. Splitting the two causes needs its own
            # column and is worth doing only if the band turns out to be
            # dropping rules anyone wanted.
            suppressed=len(hits) - len(fresh),
        )
        if fresh:
            # `rule_ids` is `fresh`, i.e. AFTER exclude_rule_ids. A rule the
            # session already holds was considered and not shown, and counting
            # it would inflate the denominator with claims the agent never saw
            # — which reads as a precision problem this arm does not have.
            record_rule_surfaced(
                user_id=user_id, rule_ids=rule_ids, source="write_path_rule",
            )
    except Exception:
        logger.debug("write-path rule arm failed", exc_info=True)

    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],
        "rule_ids": rule_ids,
    }


async def build_tool_rule_hint(
    user_id: int,
    tool_name: str,
    command: str,
    *,
    project_id: int = 0,
    exclude_rule_ids: list[int] | None = None,
    held_rule_ids: list[int] | None = None,
) -> dict:
    """Standing rules that may apply to the ACTION about to be taken (#3476).

    The sibling of the write-path rule arm, and the surface that was missing.
    That arm is keyed on `code or path`, so a rule can only be retrieved at the
    moment of a code WRITE. Every rule about which tool to reach for — don't
    curl the forge, don't stand up a stack, don't run the suite locally, don't
    branch — was therefore unreachable at the moment it mattered, and residency
    in the always-on preload was the only surface it had.

    WHY A MECHANICAL TRIGGER AND NOT AN INSTRUCTION. Note #3089's finding is
    that a reflex generates no query: you reach for `curl` confidently, with no
    moment of doubt, so any surface that waits to be asked never fires. Here
    nothing has to be asked — the tool call IS the query, and the reflex has to
    become a tool call before it can do anything.

    Deliberately TOOL-AGNOSTIC: takes a name and a string. The hook decides
    which tools it watches, so widening the matcher is a `hooks.json` edit with
    no change here.

    EVERY TIER, since #3702 — the tier filter this docstring used to describe
    is gone from both arms, for the reason recorded at RULEHINT_LIMIT: present
    in context and salient at the moment are different properties, and only the
    second is what this arm is for.

    Fails open and returns an empty context on any error: a recall aid may
    never break the operator's action.
    """
    out: dict = {"context": "", "rule_ids": []}
    command = (command or "").strip()
    if not command:
        return out

    try:
        cfg = await get_writepath_config(user_id)
        if not cfg.get("enabled"):
            return out

        # The command text is the query. A long heredoc or a pasted script
        # would otherwise push the meaningful head of the command out of the
        # embedding window, so it is bounded — the verb and its target sit at
        # the front, which is the part a rule is about.
        query = command[:_TOOL_QUERY_CHARS]

        t0 = time.perf_counter()
        _rep_ptr: dict = {}
        hits = await semantic_search_rules(
            user_id, query, limit=RULEHINT_LIMIT,
            threshold=cfg["tool_rule_threshold"],
            report=_rep_ptr, project_id=project_id or None,
        )
        duration_ms = (time.perf_counter() - t0) * 1000.0

        already = set(exclude_rule_ids or [])
        held = set(held_rule_ids or [])
        # Band first, dedup second — see the sibling arm for why that order is
        # load-bearing rather than incidental.
        kept = _rule_band(hits)
        fresh = [(score, rule) for score, rule in kept if rule.id not in already]

        # Logged BEFORE the early return, for the reason spelled out at length
        # on the write-path arm above: a call that found nothing is the only
        # evidence a threshold is too high, and an arm that logs only the calls
        # it liked reports a flawless clear-rate however badly it is tuned.
        # This arm shipped with the same defect inherited from its sibling, and
        # it mattered more here — a surface with no rows at all cannot be told
        # apart from a hook that never fired, which is precisely the silent
        # failure the arm was built to stop.
        record_retrieval(
            user_id=user_id, source="pre_tool_rule", query=query,
            threshold=cfg["tool_rule_threshold"], limit=RULEHINT_LIMIT,
            project_id=project_id,
            is_task=None, results=fresh, duration_ms=duration_ms,
            best_available=_rep_ptr.get("best_available_score"),
            best_available_id=_rep_ptr.get("best_available_id"),
            searched=bool(_rep_ptr.get("searched", True)),
            # See the sibling arm, including why this now counts BOTH the
            # ledger and the band. It matters more here: this arm fires on
            # every Bash call, so a long session excludes its way to an
            # all-zero row and the threshold looks wrong when nothing about it
            # is.
            suppressed=len(hits) - len(fresh),
        )
        # `kept`, not `fresh` (#3750). A call whose only hit is a repeat still
        # has something to say — the arm just says it differently.
        if not kept:
            return out

        lines = [
            _rule_hint_line(
                rule, where=f"to this {tool_name} call",
                seen=rule.id in already,
                held=rule.id in held,
                # Rank decides volume (#3851): the ranker's best guess gets the
                # trigger, the rest get cited.
                compact=idx > 0,
            )
            for idx, (_score, rule) in enumerate(kept)
        ]
        # FRESH-ONLY, for the reason given on the sibling arm: a reference is a
        # rendering decision, not a retrieval outcome, and counting it here
        # would inflate the denominator pull_through is read from.
        rule_ids = [rule.id for _score, rule in fresh]

        # RANKED, not ambient: this arm chose what it showed, so a pull can
        # settle whether the choice was any good. `rule_usage.RANKED_SOURCES`
        # carries the same name.
        #
        # GUARDED, which it did not need to be before #3750: `fresh` can now be
        # empty on a call that still emitted a line, and recording a surfacing
        # of nothing would write an event with no rules in it.
        if rule_ids:
            record_rule_surfaced(
                user_id=user_id, rule_ids=rule_ids, source="pre_tool_rule",
            )
        out["context"] = "\n".join(lines)
        out["rule_ids"] = rule_ids
    except Exception:
        logger.debug("pre-tool rule arm failed", exc_info=True)
    return out


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)"
        # What renders a css family (milestone 302): the consumer count is
        # the datum that separates a shared recipe from a scoped convention.
        cons = f.get("consumers")
        if cons is not None:
            n_t = cons.get("count", 0)
            used = f"; used by {n_t} template{'s' if n_t != 1 else ''}"
            if cons.get("paths"):
                used += ": " + ", ".join(f"`{x}`" for x in cons["paths"])
                extra = n_t - len(cons["paths"])
                if extra > 0:
                    used += f" +{extra} more"
            files += used
        # 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."
    )




def _goal_line(goal: str, project_id: int) -> str:
    """The Goal line, trimmed at a word break with a visible cut.

    A raw slice ended mid-word with nothing to say more existed, so a reader
    took half a sentence for the whole goal (#4036).
    """
    if not goal:
        return ""
    flat = " ".join(goal.split())
    if len(flat) <= _GOAL_CHARS:
        return f"Goal: {flat}"
    short = textwrap.shorten(flat, width=_GOAL_CHARS, placeholder="…")
    if short == "…":  # one unbroken word longer than the cap
        short = flat[: _GOAL_CHARS - 1] + "…"
    return f"Goal: {short} (full goal: `enter_project({project_id})`)"


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, or from a `.scribe`
            marker file naming the project directly — the only key a session
            outside a git repo has (#4085). A non-zero id that does not
            resolve is reported rather than silently dropped.
        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, "project": dict | None}.

    It carried `rule_count` and `rules_etag` until milestone 394, when the
    preload it described was removed. The etag let the hook hand a marker back
    on each write so the server could say whether the resident rules had
    moved; nothing is resident now, so nothing can have moved, and a rule is
    re-retrieved at the moment it applies rather than held and aged.
    `context` is markdown an adapter can drop into its session verbatim; it
    is capped at _MAX_CHARS with an explicit truncation note.

    LIVE STATE ONLY (decision #4027, milestone 410). This used to open with the
    rules reflex and close with a recall reflex — a fifth copy of guidance the
    using-scribe skill owns, arriving in every session beside the other four.
    It now says only what the server alone knows about THIS session: the
    active project, its open work, its design system, or that the working repo
    is unbound. How to work with Scribe is the skill's to say, and it says it
    once.
    """
    lines: list[str] = ["# Scribe — live session state"]

    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})",
                _goal_line(goal, project.id),
                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 "")
                        + ".",
                        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), inherited house style included: "
                        f"`get_design_system({design['id']})` → "
                        f"`resolved_guidance`.",
                    ]
    # Nothing loaded — say which nothing (#4085). This used to hang off the
    # `if project_id:` above as an `elif`, which meant an id that was SENT and
    # did not resolve produced no message at all: the outer branch was taken,
    # the inner one was not, and the caller got a context that simply omitted
    # the project it had asked for. That is the one case worth being loudest
    # about, because the caller is holding a pointer it believes in.
    if project_dict is None:
        if project_id:
            lines += [
                "",
                f"## Project {project_id} could not be loaded",
                f"This session asked for project {project_id}, but this account "
                "cannot read it — the id may belong to a different Scribe "
                "instance, or the project may have been deleted. "
                "`list_projects` shows what is readable here.",
            ]
        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.",
            ]
        else:
            lines += ["", "No Scribe project is bound to this working directory."]

    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)"

    return {
        "context": context,
        "project": project_dict,
    }