"""Session-context rendering for the Grimoire plugin's SessionStart hook. The plugin's hook curls `GET /api/plugin/context` at session start and injects the returned text as `additionalContext`, giving Grimoire a push channel into the session. This module renders that text. Design note — altitude: we inject rule *titles* grouped by topic (a compact index), NOT every rule's full statement. A working rulebook's 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 ("Run migrations before deploying"), so the index alone already steers behavior. """ from __future__ import annotations import logging import re import textwrap from grimoire.services import design_systems as design_systems_svc from grimoire.services import knowledge as knowledge_svc from grimoire.services import notes as notes_svc from grimoire.services import projects as projects_svc from grimoire.services import record_refs as record_refs_svc from grimoire.services import shape_ledger as shape_ledger_svc from grimoire.services import snippets as snippets_svc from grimoire.services import task_claims as task_claims_svc from grimoire.services.access import label_shared_items, owner_names_for from grimoire.services.embeddings import ( document_title, semantic_search_notes, semantic_search_rules, ) from grimoire.services import lesson_rules as lesson_rules_svc from grimoire.services.lessons import LESSON_NOTE_TYPE from grimoire.services.note_usage import record_surfaced from grimoire.services.rule_usage import record_rule_surfaced from grimoire.services.supersession import superseded_ids from grimoire.services.retrieval_surfaces import ( MAX_BUDGET, SURFACES, budget_for, floor_for, ) from grimoire.services import retrieval_pipeline as rp from grimoire.services.retrieval_pipeline import ( # noqa: F401 - re-exported _RULEHINT_BAND, _rule_band, _rule_hint_line, checkpoint_for, checkpoint_reason, VIA_LESSON_LIMIT, _menu_label, _menu_name, _menu_passage, _menu_seen_line, _record_kind, menu_entry, ) from grimoire.services.retrieval_telemetry import record_retrieval from grimoire.services.settings import get_setting from grimoire.services.systems import system_names_for from grimoire.services.text import elide from grimoire.services import system_rulings as system_rulings_svc logger = logging.getLogger(__name__) # Defensive cap below Claude Code's 10k additionalContext limit. _MAX_CHARS = 9000 # The notes renderer — a record's name, kind, System and matched passage — # lives in `retrieval_pipeline` with the rest of the notes stages and is # re-exported above. # How much of a named record's body sits under its line. The opening, not an # elided middle: nothing matched here, so there is no passage to keep, and what # the reader needs is what the record IS — which is where a record says it. _NAMED_OPENING_CHARS = 240 def _named_opening(body: str | None) -> str: text = " ".join((body or "").split()) if len(text) <= _NAMED_OPENING_CHARS: return text return text[:_NAMED_OPENING_CHARS].rsplit(" ", 1)[0] + " …" async def _named_records(user_id: int, ids: list[int]) -> list: """The records behind ids the operator named that this user may open. An id that is trashed, someone else's private record or not a record at all is dropped without a word: the parser cannot tell "4448" the task from "4448" a quantity it misread, and the lookup is what settles it. """ found = [] for nid in ids: got = await notes_svc.get_note_for_user(user_id, nid) if got is not None and got[0].deleted_at is None: found.append(got[0]) return found # 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. _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. # # The defaults below are STARTING POINTS, and correcting them is the model's # job, not the operator's: `retrieval_surfaces` says what is in force, # `retrieval_telemetry(near_miss_samples=N)` says what it refused, and # `tune_retrieval` moves it with the reason attached. The operator can set any # of them in Settings and their change is recorded the same way — but nobody # has to read a log to get correct behaviour out of this. AUTOINJECT_ENABLED_KEY = "kb_autoinject_enabled" # The key and the value both live in the registry; these names exist because # the comments beside them record why each number is what it is, and because # the Settings UI agreement test pairs them with the Vue refs by name. AUTOINJECT_THRESHOLD_KEY = SURFACES["auto_inject"].floor_key AUTOINJECT_TOP_K_KEY = SURFACES["auto_inject"].budget_key AUTOINJECT_DEFAULT_ENABLED = True AUTOINJECT_DEFAULT_THRESHOLD = SURFACES["auto_inject"].floor_default AUTOINJECT_DEFAULT_TOP_K = SURFACES["auto_inject"].budget_default # The write-path trigger gets its own on/off switch, its own threshold, and # shares only top-k. Its threshold differs because the semantic arm's scores # for CODE sit far above what the same threshold means for PROSE (probe # payloads against a recorded helper): # 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. The margin gate can't rescue it either: the notes band (retrieval_pipeline._NOTE_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, which is tuned on prose. # Tune from retrieval_logs (source='write_path') and note_usage pull-through; a # cross-encoder rerank would subsume this bump. WRITEPATH_ENABLED_KEY = "kb_writepath_enabled" WRITEPATH_THRESHOLD_KEY = SURFACES["write_path"].floor_key WRITEPATH_DEFAULT_ENABLED = True WRITEPATH_DEFAULT_THRESHOLD = SURFACES["write_path"].floor_default # The standing-rule arm gets its own bar, for a third corpus: 0.68 above is # calibrated on code against note PROSE, not code against RULE TEXT. # # THE STRUCTURAL ARGUMENT, which holds 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. (The arms search every tier, which # makes the pool larger; the bar still errs toward silence on purpose.) # 2. Rules are short imperative technical English — a far more HOMOGENEOUS # corpus than note prose, and a more homogeneous corpus has a HIGHER # noise floor than prose's 0.55-0.63. # # 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 Grimoire 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. # # TUNE IT FROM YOUR OWN INSTANCE: `retrieval_telemetry` reports # `rule_usage.pull_through`. Raise this if rules arrive unread; lower it if # rules you needed never arrived. What would RETIRE it: a cross-encoder rerank, # which would make a similarity bar the wrong control entirely. # # This bar is the WRITE-PATH arm's. A code payload is long and rich, which is # the case 0.72 suits; the command arm has its own bar below. RULEHINT_THRESHOLD_KEY = SURFACES["write_path_rule"].floor_key RULEHINT_DEFAULT_THRESHOLD = SURFACES["write_path_rule"].floor_default # THE COMMAND ARM'S OWN BAR, AND WHY IT IS NOT THE WRITE PATH'S. # # 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 the write path's 0.72 the command arm # turns nearly everything away, with its refused mass piled just under the # line, which is the shape a bar set too high leaves. # # WHAT 0.68 IS CALIBRATED AGAINST. Replayed queries, consequential acts # against innocuous ones, scored against a rule corpus with symptom-bearing # triggers: # # 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 would retrieve nothing, though # the right rules ranked first for each. # # THE SEPARATION IS NARROW (about 0.01), and that is a caveat, not a result. # Eight probes set a direction; they do not settle a number. # `near_miss_samples` on real traffic is what settles it, and this is the bar # to re-read first. # # WHY A LOWER BAR IS SAFE HERE. The band 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 innocuous scores near the line are largely CROSS-PROJECT BLEED rather # than bad ranking: another project's shell-command rule matching a shell # command in this one. Project scoping of rules lowers that ceiling. TOOLRULE_THRESHOLD_KEY = SURFACES["pre_tool_rule"].floor_key TOOLRULE_DEFAULT_THRESHOLD = SURFACES["pre_tool_rule"].floor_default # A SET OF RULES PER ACT, NOT THE SINGLE BEST ONE. # # Nothing is preloaded, so this arm is the whole delivery, and an act is often # governed by several rules at once — a push can be bound by where to push, # what never to push to, polling CI and letting one action land before the # next, and each alone permits the mistake the others catch. "The single best" # is not a coherent answer when four rules bind. # # WHY A CAP PLUS A BAND, RATHER THAN A BIGGER CAP. A fixed k invents lines — it # fills slots whether or not anything deserves them, and over a small corpus # the second slot is usually the second-best noise. 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 is a ceiling on the worst case, not the usual answer. It is a # DEFAULT: both rule act-arms read their own budget from the registry, so this # is the value an install starts at, not the value it is stuck with. RULEHINT_LIMIT = SURFACES["write_path_rule"].budget_default # ── The pre-act checkpoint ─────────────────────────────────────────────── # # WHAT A CHECKPOINT IS, AND WHY IT IS NOT A LOUDER HINT. Every arm above # appends text to an act the agent has already composed. Claude Code delivers # `additionalContext` alongside the tool result, so by the time the line is # read the call is written and the rule reads as commentary on a decision # already made. # # A checkpoint is the same retrieval, spent differently: the hook returns a # `deny` and the act does not run, so the rule's own text is read BEFORE the # call exists. The remedy is `get_rule(id)` and nothing else — which is why # the condition below is "the session has not opened it" rather than "the # session has not recorded an outcome for it". An outcome can be satisfied # with one cheap call that asserts compliance without producing any, and a # checkpoint that can be dismissed that way manufactures exactly the # compliance data the outcome telemetry exists to measure. Reading a rule cannot be faked in # that direction: after `get_rule` the statement is in context, which is the # whole of what was wanted. # # WHAT "CONSEQUENTIAL" MEANS HERE, AND WHY IT IS NOT A LIST. The obvious # implementation enumerates act kinds — a write to product code, a schema # change, a bulk classification, a merge. Those are consequential only where # an operator wrote rules about them, and a shipped list would hard-code one # install's corpus into the product. So the corpus decides: an act is consequential when the install's own rules speak to it # with high confidence. A fresh install with no rules never stops anything, # and an install whose rules are about something else entirely stops on that # instead. _CHECKPOINT_THRESHOLD_KEY = "kb_checkpoint_threshold" # 0.80 is meant to sit above the p90 of what both rule act-arms return and # below their max: it selects from the top decile of an already-selective arm # rather than from its bulk, and it stays reachable. On a heavily used install # that is on the order of ten stops a day across all sessions. # # A SETTING, because the number above is a distance in one embedding model's # geometry over one corpus and cannot transfer (retrieval_surfaces' opening # argument). The default ships as a starting point with the means to correct # it, which is the only honest form for a number like this. _CHECKPOINT_DEFAULT = 0.80 # A session cannot be stopped more than this many times, however the corpus # scores. Not a tuning value — a guard on the worst case, like MAX_BUDGET: a # mis-set floor or a corpus that suddenly resembles everything must degrade to # a noisy session, never to one that cannot make progress. The hook enforces # it, because only the hook knows what a session is. CHECKPOINT_SESSION_CAP = 5 # AND A REPEAT COMPETES ON RANK ALONE, WHICH SURVIVES THE BAND. # # A hit already on the session's exclusion ledger is RENDERED rather than # dropped. When the top-ranked hit is one the session has already seen, it # keeps its place, and nothing is promoted past it: 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. # # 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 be withholding, 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 PAID BY RENDERING, NOT WISHED AWAY. # A full line runs to a hundred-odd tokens once the trigger is rendered, so # five full lines before EVERY Bash call is several hundred tokens. So the # set pairs with a compact rendering (see `_rule_hint_line`): top hit full, # the rest as references, about half the cost of five full lines. A voice # that speaks at full volume every time is where a reader stops reading; # the answer is quieter later lines, not fewer rules. # # The reference keeps its TAIL and loses only its trigger, which is both the # cheaper and the safer cut — see `_rule_hint_line`. # # 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, not to # a rule that ranks first being quietly demoted for having won before. # WHY THE ARMS SEARCH EVERY TIER. # # Filtering out always-on rules, on the reasoning that they are already in # the session, 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. A tier filter # would make 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 is in that class. # # And such a filter does 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. # # The risk is real — a hint that fires on every write and says obvious things # teaches the reader to skip the block. `retrieval_logs` records top_score, # result_count and the query for every call, so it is measurable: rules # clearing the bar often and at high scores mean the WORK IS THE BAR, not a # tier filter. # How much of a command reaches the embedding. 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.example.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. # # 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. # A flat floor rather than a length-to-threshold curve, and it saves a # pointless embedding round-trip on trivial edits. WRITEPATH_MIN_CODE_CHARS = 48 # --- concept extraction for the semantic arm's query ------------------------- # 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. # # Querying it with raw code and no prose matches poorly. Measured against one # recorded snippet: # 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. 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. 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 the raw payload. _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 # Hard ceiling on top-k regardless of the user's setting — this is an # awareness menu, never a content dump. The ceiling is shared by every # surface: it bounds the same thing everywhere — how many lines a single # unsolicited injection may occupy. _AUTOINJECT_MAX_TOP_K = MAX_BUDGET # THE CONVERSATION BESIDE THE PROMPT. The operator's message alone is a weak # query mid-session: it is mostly a follow-up — "yes do that", "now fix the # filter" — that names nothing a record could match. The hook sends the tail of the last assistant reply as `context`, and it is # appended to a SHORT prompt only: a prompt that already says what it is about # is the better query on its own, and diluting it is the one way this can make # retrieval worse. # # Neither number costs a model token. The query is embedding input; what the # session pays for is the menu, and the budget bounds that unchanged. # 280 — about two sentences. Above it a prompt carries its own subject. # 600 — the context cap. Prompt + context stays well inside bge-small's # 512-token window, prompt FIRST, so truncation can only ever cut # context and never the operator's words. _AUTOINJECT_CONTEXT_PROMPT_MAX = 280 _AUTOINJECT_CONTEXT_MAX = 600 def _autoinject_query(prompt: str, context: str) -> str: """The prompt, with recent conversation appended when the prompt is thin. The prompt leads and context is its tail, cut from the END of the reply — a reply's closing lines are where it says what it did and what is next, which is what the operator's follow-up is answering. """ ctx = " ".join((context or "").split())[-_AUTOINJECT_CONTEXT_MAX:] if not ctx or len(prompt) > _AUTOINJECT_CONTEXT_PROMPT_MAX: return prompt return f"{prompt}\n\n{ctx}" # --- the prompt-boundary rule arm ------------------------------------------- # # 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. PROMPTRULE_THRESHOLD_KEY = SURFACES["prompt_rule"].floor_key # INHERITED FROM THE ACT ARMS. 0.72 is 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`, so # real traffic settles it — read `near_miss_samples` before moving this, not # the percentile alone. PROMPTRULE_DEFAULT_THRESHOLD = SURFACES["prompt_rule"].floor_default # ANCHORED ON THIS HOOK'S BUDGET rather than the act arms'. 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 often contains more than one act: "merge to main and then start on # X" is two, governed by different rules. PROMPTRULE_LIMIT = SURFACES["prompt_rule"].budget_default 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 (grimoire_sync_processes.sh) writes one ~/.claude/skills/grimoire-proc-/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 Grimoire 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. 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 -). """ 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 Grimoire 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 Grimoire 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 Grimoire has (see SCOPE above): a # Process installed as a skill and never once opened must be # distinguishable from one never installed. 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}. THE TWO NUMBERS COME FROM THE REGISTRY, like every arm's: a tool is expected to MOVE them, and a tuning surface cannot be consistent across arms that each spell their configuration differently. `enabled` stays here: it is this surface's own switch, not a tunable number, and the registry deliberately holds only the pair a floor-tuner touches. """ 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") return { "enabled": enabled, "threshold": await floor_for(user_id, "auto_inject"), "top_k": await budget_for(user_id, "auto_inject"), } async def build_autoinject_hint( user_id: int, query: str, project_id: int = 0, exclude_ids: list[int] | None = None, context: str = "", ) -> dict: """Title-first awareness hint for the plugin's UserPromptSubmit hook. The four anti-bloat gates: 1. high-confidence threshold (stricter than pull) — set per-user; 2. margin gate — keep only hits within retrieval_pipeline._NOTE_BAND of the top score; 3. session marking — caller passes already-injected ids as `exclude_ids` and they are rendered again with `[seen]`, never withheld; 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 # RECORDS NAMED BY NUMBER, read from the operator's own words before # the query is enriched: a number in the assistant's reply is not one the # operator named. "go ahead with 4448" says which record is meant, and a # number means nothing to an embedding, so these are looked up rather than # left to a ranking that can only match the words around them. named = await _named_records(user_id, record_refs_svc.named_record_ids(q)) named_ids = [int(n.id) for n in named] # Everything below searches, logs and fills its slots on the ENRICHED # query, so the telemetry row records what was actually asked. q = _autoinject_query(q, context) # THE LEDGER IS A RENDERING FACT, NOT A RETRIEVAL ONE. Excluding seen ids # from the search would make the second time a note is the best answer # SILENCE, indistinguishable from "nothing matched", and would measure # `best_available` against a candidate set the caller had already edited. # So every repeat is still ranked, still shown, and carries a marker saying # it was surfaced before. A note line is a title and a score — the repeat costs about as # much as the comma in this sentence — so there is nothing here to save. already = {int(i) for i in (exclude_ids or [])} # The ranked half is `retrieval_pipeline.run_note_arm`: # search, the fresh/repeat split, the call row, the band, the reuse and # lesson slots in that order, and the surfacing rows are written there. # What is decided HERE is what makes this moment this moment — the query # the operator's words became, and the records they named by number. result = await rp.run_note_arm( rp.AUTO_INJECT, rp.NoteMoment( user_id=user_id, query=q, project_id=project_id, seen=frozenset(already), named=frozenset(named_ids), ), floor=cfg["threshold"], budget=cfg["top_k"], io=_note_io(), ) kept = result.menu if not kept and not named: return empty # 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 n in [*named, *(n for _s, n in kept)] if n.user_id != user_id }) # A superseded record is DEMOTED, not removed — `menu_entry` says # so on the line. One query for the whole menu. stale = await superseded_ids([*named_ids, *(int(n.id) for _s, n in kept)]) systems = await system_names_for( {i for i in named_ids if i not in already} | {int(n.id) for _s, n in kept if int(n.id) not in already} ) def _shared_by(note) -> str: if note.user_id == user_id: return "" return owners.get(int(note.user_id)) or "another user" lines: list[str] = [] note_ids: list[int] = [] # FIRST, because the operator said which record they meant and everything # after this block is a guess at what else might bear on it. The header # says these were found by NUMBER: the parser reads "with 4448" as a # reference, and the reader is the one placed to notice it was a quantity. if named: lines.append( "> Named in your message — the records with those numbers, looked " "up by id rather than matched. Open any in full with `get_note(id)`; " "a line marked `seen` is already in your context:" ) for note in named: nid = int(note.id) note_ids.append(nid) # The OPENING under the line, not a passage: nothing matched here, so # there is no passage to keep, and what the reader needs is what the # record IS — which is where a record says it. lines.extend(menu_entry( nid, kind=_record_kind(note), name=_menu_name(note.title, note.note_type, note.data, note.body), systems=systems.get(nid), seen=nid in already, stale=nid in stale, shared_by=_shared_by(note), under=_named_opening(note.body), )) # "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. # A repeat is shown again with a marker rather than withheld, so the header # says what the marker means, once, rather than each repeated line having # to explain itself. if kept: lines.append( "> Possibly relevant from your Grimoire records — open any in full with " "`get_note(id)`, or `get_snippet` / `get_process` / `get_lesson` for " "those kinds. Each line is a record's name, its kind and System, and " "the passage that matched; a line marked `seen` is a pointer to one " "already shown this session, so it is in your context:" ) # THE REGISTER, SAID ONCE AND ONLY WHEN IT APPLIES. # # A lesson does not always have to be followed, and the risk is not that a # reader mistakes a lesson for a rule — this menu's voice is already the non-binding one, deliberately # (see the `seen` marker below). The risk is the opposite: read as one more # title in a list of MATERIAL, a lesson looks like something to open if # curious, when it is advice someone paid for. The line has to say "weigh # this" without acquiring the rule arms' "before deciding it does not # apply", which binds. # # In the HEADER rather than on each line, for the reason the `seen` flag is # a flag: the meaning is the same for every lesson on the menu, and a # clause repeated per line would cost more than it says. Conditional # because a menu with no lesson should not pay for the sentence, and # because a header that explains an absent kind reads as boilerplate — # which is how a reader learns to skip headers. if any(_record_kind(n) == LESSON_NOTE_TYPE for _s, n in kept): lines.append( "> A line marked `lesson` is something an earlier session learned " "the hard way, kept because it should transfer. Weigh it against " "what you are doing and use your judgement — a lesson is not a " "rule and binds nothing." ) for score, note in kept: nid = int(note.id) note_ids.append(nid) # The NAME, not the title: a snippet's or lesson's title is its # embedding shape, trigger and all, and can run past 1,500 characters. name = _menu_name(note.title, note.note_type, note.data, note.body) # The passage that earned the line, WHOLE, indented under it. # Absent when the record has no stored chunk — an un-embedded row, or # the reserved lesson and reuse slots, which are fetched by their own # queries and so are not in this search's report. No fallback to the # body's opening: on a menu that would be a line of preamble dressed as # a reason, and a reader cannot tell the two apart once indented alike. passage = "" if nid in already else _menu_passage( document_title(note.title, note.note_type, note.data, note.body), (result.chunks.get(nid) or {}).get("text"), name, ) who = _shared_by(note) lines.extend(menu_entry( nid, kind=_record_kind(note), name=name, systems=systems.get(nid), seen=nid in already, stale=nid in stale, score=score, shared_by=f"{who}, treat as a suggestion" if who else "", under=passage, )) # A lookup, not a ranking, so it writes no retrieval_logs row — there is no # score or bar to tune — and its surfacings are its own source, which is # what lets pull-through say whether a named record gets opened. The # ranked menu's own rows were written by the pipeline. named_fresh = [i for i in named_ids if i not in already] if named_fresh: record_surfaced( user_id=user_id, note_ids=named_fresh, source="named_ref", project_id=project_id, ) # The lessons this menu put in front of the reader, repeats included — for # the soft-link recorder, which pairs them with the rule arm's # lines in the same response. Relevance, not the session ledger, is what # makes a co-arrival, so a `seen` lesson counts — and so does one the # operator named, which is relevance by their own say-so. lesson_ids = [ int(n.id) for n in [*named, *(n for _s, n in kept)] if _record_kind(n) == LESSON_NOTE_TYPE ] return { "context": "\n".join(lines), "note_ids": note_ids, "config": cfg, "lesson_ids": lesson_ids, } def _note_io() -> rp.NoteIO: """The ranker and recorders the notes arms report to, read at call time — `_rule_io`'s reason: whatever this module's names are when the arm runs.""" return rp.NoteIO( search=semantic_search_notes, record_retrieval=record_retrieval, record_surfaced=record_surfaced, ) async def _rules_via_lessons( user_id: int, query: str, *, project_id: int | None, skip: set[int], held: set[int], where: str, ) -> tuple[list[str], list[int]]: """Rule lines reached through a matching lesson's CONFIRMED links — the pipeline's via-lesson arm (`retrieval_pipeline.run_via_lesson_arm`, where the design is written), with its I/O read from this module at call time. The lesson bar is the notes menu's own threshold. Returns (lines, rule ids shown). """ async def bar() -> float: return (await get_autoinject_config(user_id))["threshold"] result = await rp.run_via_lesson_arm( rp.RuleMoment( user_id=user_id, query=query, project_id=project_id, where=where, held=frozenset(held), ), skip=frozenset(skip), io=rp.ViaLessonIO( search=semantic_search_notes, linked=lesson_rules_svc.confirmed_lessons, rules_for=lesson_rules_svc.confirmed_rules_in_scope, floor=bar, record_retrieval=record_retrieval, record_rule_surfaced=record_rule_surfaced, ), ) return result.lines, result.rule_ids async def _rule_moment( arm: rp.RuleArm, moment: rp.RuleMoment, *, floor: float, budget: int, checkpoint_floor: float = 0.0, ) -> rp.RuleResult: """One rule moment, composed: the direct arm, then the rules its matching lessons bring in — after the band and never in place of it. Every rule builder below calls this ONE function. The via-lesson step searches with the query the arm searched with, and skips what the arm named plus what the session's ledger holds — suppression is by RULE, whichever lesson reached it. Its lines and fresh ids join the arm's; `shown_rule_ids` stays the direct arm's band, which is what the soft-link recorder pairs. """ result = await rp.run_rule_arm( arm, moment, floor=floor, budget=budget, io=_rule_io(), checkpoint_floor=checkpoint_floor, ) lines, ids = await _rules_via_lessons( moment.user_id, moment.query, project_id=moment.project_id or None, skip=set(moment.exclude) | set(result.shown_rule_ids), held=set(moment.held), where=moment.where, ) result.lines.extend(lines) result.rule_ids.extend(ids) return result 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, context: str = "", ) -> dict: """Rules and preferences that may apply to what the operator just asked. The third rule arm, and the one that reaches what the other two cannot. `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; nothing is preloaded, so without this arm that half of the corpus would never arrive. 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, and the two share no header. 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. NO `checkpoint` KEY: the act arms can hold a call because there is a composed act to hold, while this one fires before anything has been decided. 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 # The same enrichment the notes arm gets. A rule is meant to # arrive while the work it governs is under way, not once the operator # names it — and "yes, go ahead" names nothing a trigger can match, while # the reply it answers ("commit this to dev and push") does. q = _autoinject_query(q, context) try: # SCOPED TO THIS SESSION'S PROJECT: global rules plus # the bound project's own. An unbound session (project_id 0) gets # global rules only — this surface speaks unasked, and a whole-rulebook # answer is only right for someone who asked the whole rulebook. result = await _rule_moment( rp.PROMPT_RULE, rp.RuleMoment( user_id=user_id, query=q, project_id=project_id, where="to this request", exclude=frozenset(exclude_rule_ids or []), held=frozenset(held_rule_ids or []), ), floor=await floor_for(user_id, "prompt_rule"), budget=await budget_for(user_id, "prompt_rule"), ) if result.lines: out["context"] = "\n".join(result.lines) out["rule_ids"] = result.rule_ids # Every rule LINE the direct arm showed, repeats and the reserved # slot included — for the soft-link recorder. Distinct # from `rule_ids`, which is telemetry's fresh-only cut. out["shown_rule_ids"] = result.shown_rule_ids except Exception: logger.debug("prompt rule arm failed", exc_info=True) return out def _rule_io() -> rp.RuleIO: """The ranker and recorders the rule arms report to, read at call time. Resolved from this module's names on every call rather than bound once, so the pipeline reports to whatever this module's `semantic_search_rules`, `record_retrieval` and `record_rule_surfaced` are at the moment it runs. """ return rp.RuleIO( search=semantic_search_rules, record_retrieval=record_retrieval, record_rule_surfaced=record_rule_surfaced, ) # --- Write-path trigger: 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. 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. # The reverse lookup answers it. # - 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. """ # The NAME, not the composed title — a snippet's title carries its whole # trigger and can run to kilobytes, again on every repeat. # `snippet` is a snippet record's field dict — but on a search hit it is # the matched excerpt, a string, under the same key. fields = item.get("snippet") title = ( item.get("name") or (fields.get("name") if isinstance(fields, dict) else None) or (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 ------------------------------------------------ # Retrieval is concept-shaped, and concepts are language-agnostic: a query about # a TypeScript union-find matches a PYTHON snippet comfortably over the bar. That # is a feature — 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. # # 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 (see the table above _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 the raw payload. """ 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 its recorded snippet 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 — and the two rule arms' numbers alongside. Three surfaces' worth of configuration arrives in one call because one hook request drives all three arms. They are still three SURFACES with three independent pairs, resolved from the registry. Each floor's reason is recorded where its default is: WRITEPATH (code embeddings sit on a much higher similarity floor than prose), RULEHINT (a third corpus again), TOOLRULE (a shell command is a different query shape from a code payload and scores lower for the same relevance). THE BUDGET IS SPLIT TOO. "How many titles at once" does not mean the same thing on both surfaces: this arm fires before every Write and Edit while auto-inject fires once a turn, so the same number buys wildly different amounts of attention. `write_path` inherits auto-inject's value when it has none of its own, so no install that tuned the shared knob loses it. """ enabled_raw = await get_setting( user_id, WRITEPATH_ENABLED_KEY, "true" if WRITEPATH_DEFAULT_ENABLED else "false", ) return { "enabled": enabled_raw.strip().lower() in ("true", "1", "yes", "on"), "threshold": await floor_for(user_id, "write_path"), "top_k": await budget_for(user_id, "write_path"), "rule_threshold": await floor_for(user_id, "write_path_rule"), "rule_top_k": await budget_for(user_id, "write_path_rule"), "tool_rule_threshold": await floor_for(user_id, "pre_tool_rule"), "tool_rule_top_k": await budget_for(user_id, "pre_tool_rule"), # NOT from the surfaces registry, and deliberately so. Everything in # that table is a pair belonging to one QUERY — a floor saying what is # worth ranking and a budget saying how many lines it may spend. The # checkpoint runs no query of its own; it re-reads hits the two rule # arms already produced and asks a different question of them. Putting # it in the registry would give it a phantom budget and make the # tuning tool offer to change how many checkpoints an act may raise, # which is not a number anyone should have. "checkpoint_threshold": await _checkpoint_floor(user_id), } async def _checkpoint_floor(user_id: int) -> float: """The confidence at which a hint becomes a stop. Clamped, never trusted. A floor read out of settings reaches here as operator-typed text. Below zero it would stop every act with a rule anywhere near it; above one it can never fire and the feature is silently dead — which is why this clamps rather than trusting a validation at the door. """ raw = await get_setting( user_id, _CHECKPOINT_THRESHOLD_KEY, str(_CHECKPOINT_DEFAULT), ) try: value = float(str(raw).strip()) except (TypeError, ValueError): return _CHECKPOINT_DEFAULT return min(1.0, max(0.0, value)) 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, seen_ruling_systems: 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: "this snippet records the file you're editing; if the edit changes the recorded shape, updating the record is part of the edit." 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 track the session 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". The two channels also ACT differently: a reuse repeat is rendered again with a `seen` marker, while the sync class still shows once, because its claim is about an edit in progress rather than about a record's continuing relevance and repeating it would be nagging. Carries auto-inject's anti-bloat gates (margin, session marking, 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 (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 ("Grimoire"), 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. Un-scored surfacing has its own home instead: 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: 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 carry it as a PROPOSAL (see shape_ledger.suggest_write_path_instances) and the result's ``suggested`` lists them. Never a verdict: the agent that wrote the code judges it at the end of the turn, shown this as evidence. The route passes it only for a caller allowed to write — a read-scoped key gets the hint, never a ledger write. ``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, "suggested": [], "divergence": [], "derive": [], "derive_keys": [], "rule_ids": [], "ruling_system_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` IS THIS CALL'S OWN MENU, and nothing else. "Already listed a few # lines above" is a reason to stay quiet; "shown at some earlier point in # the session" is not — the session ledger is a marker instead, on both # reuse arms. seen: set[int] = set() 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) # MARKED HERE TOO, not just on the semantic arm, because these are one # menu. A hint where some repeats carry `seen` and others are silently # dropped — decided by which arm happened to find them — is worse than # either rule applied consistently: the marker would read as a complete # account of what the session has met before, and it would not be one. item["seen"] = nid in excluded placed.append(("nearby · seen" if nid in excluded else "nearby", item)) # The stamping feed's "actually pulled it" half. 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. # 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. 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 # Declared out here because the search below is conditional — this arm runs # only when the menu has room AND a query survived the floor. The render # loop is not conditional, so it needs something to read either way, and an # empty mapping means every line falls back to its title alone. wp_chunks: dict[int, dict] = {} if remaining > 0 and query: # The ranked half is `retrieval_pipeline.run_note_arm` with the # WRITE_PATH spec: which kinds it asks for and why, the withheld-menu # accounting, the fresh/repeat split, the # call row and the surfacing rows are written there. What is decided # HERE is the menu above it — what place and sync already listed is # withheld, and a PULLED snippet among those is still scored, because # its resemblance to this payload is the stamping feed's evidence. result = await rp.run_note_arm( rp.WRITE_PATH, rp.NoteMoment( user_id=user_id, query=query, project_id=project_id, seen=frozenset(excluded), in_menu=frozenset(seen), still_scored=frozenset(pulled), ), floor=cfg["threshold"], budget=remaining, io=_note_io(), ) wp_chunks = result.chunks resembles = { int(note.id): float(score) for score, note in result.answered if int(note.id) in pulled } for score, note in result.menu: # 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) marker = ( f"similar {score:.2f}" if kind == "snippet" else f"similar {score:.2f} · {kind}" ) # The repeat marker rides the same dotted list as the kind, so a # reference costs four characters and needs no second line. # Same word as the auto-inject menu deliberately: a # reader meeting `seen` on two different surfaces should not # have to work out whether they mean the same thing. if int(note.id) in excluded: marker += " · seen" scored.append(( marker, { "id": int(note.id), "title": note.title, "user_id": note.user_id, # The name the line shows, and whether this session has # it already — carried as data for the reason `kind` is: # the line is built from facts, not from # re-reading its own marker. "name": _menu_name(note.title, note.note_type, note.data, note.body), # What its chunks are prefixed with, for stripping. "doc_title": document_title( note.title, note.note_type, note.data, note.body, ), "seen": int(note.id) in excluded, # Carried, not re-read off the rendered marker. The # marker is prose assembled for a human and it already # varies by kind, language and the `seen` flag — a # header that decided what to say by matching substrings # in it would break the next time a marker is reworded, # silently and in the direction of saying nothing. "kind": kind, # Carried so the line can disclose a cross-language hit. # 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 suggestion runs whether or not anything is rendered — after dedup, # the common case is a silent hint and a pulled canon being instantiated. suggested: list[dict] = [] if stamp_shapes and pulled: try: suggested = await shape_ledger_svc.suggest_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 suggestion failed", exc_info=True) # The in-band button-B check: 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, suggested, code or "", ) except Exception: logger.warning("write-time divergence check failed", exc_info=True) # The in-band DERIVE check: 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] = [] # The standing-rule arm further down is deliberately left on the far side # of the guard below: it runs a SEMANTIC search, and moving it above would # run an embedding query on every write in the session. # The design arm is decided HERE, above the guard, so a UI write # that matched no prior art still carries it — and it returns on its own # rather than joining the guard's condition, because joining it would let # a design line switch the standing-rule arm below on for writes where it # has never run, moving that arm's call distribution under its floor. design_text, design_dedup = await _design_arm( user_id, project_id, path, set(exclude_derive or []), ) # The rulings arm, decided here for the design arm's # reasons: a lookup by path, so a write that matched no prior art still # carries its area's rulings, and it never switches the ranked arms on. rulings = await system_rulings_svc.rulings_for_paths( user_id, project_id, [path], seen=seen_ruling_systems, source="rulings_write_path", ) if not staleness and not synced and not menu and not suggested and not divergence and not derive: if design_text or rulings["lines"]: return { **empty, "context": "\n".join(rulings["lines"] + ([design_text] if design_text else [])), "derive_keys": [design_dedup] if design_dedup else [], "ruling_system_ids": rulings["system_ids"], } return empty owners = await owner_names_for({ int(it["user_id"]) for it in synced + [it for _m, it in menu] if it.get("user_id") is not None and int(it["user_id"]) != user_id }) def _owner_of(item: dict) -> str | None: owner_id = item.get("user_id") if owner_id is None or int(owner_id) == user_id: return None return owners.get(int(owner_id)) or "another user" target_lang = _language_for_path(path) rendered: list[tuple[dict, str, str | None, str]] = [] for marker, item in menu: rendered.append((item, marker, _owner_of(item), _foreign_language(item, target_lang))) lines: list[str] = list(staleness) # Rulings and the design system come first: they BIND, where everything # below is prior art. lines.extend(rulings["lines"]) if design_text: lines.append(design_text) sync_note_ids: list[int] = [] if synced: # The sync framing. 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 Grimoire 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 Grimoire for `{path}` — open one with " "`get_snippet(id)` for a snippet, `get_task(id)` for an issue, " "`get_lesson(id)` for a lesson, `get_note(id)` otherwise. Reuse a " "snippet rather than writing a fresh one-off; read an issue before " "repeating what it records " "(each line is a record's name and kind, with the passage that " "matched; a line marked `seen` is a pointer to one already shown " "this session, so it is in your context):" ) # The same clause the prompt menu carries, on the same condition and for # the same reason: this menu's three other kinds are all things that WERE # done here, and a lesson is the one line that is advice. Said once, only # when one is on the menu. if any(i.get("kind") == LESSON_NOTE_TYPE for i, _m, _o, _l in rendered): lines.append( "> A line marked `lesson` is something an earlier session learned " "the hard way, kept because it should transfer. Weigh it against " "the code you are about to write and use your judgement — a lesson " "is not a rule and binds nothing." ) # 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)) # Only the semantically-matched lines carry a passage. The records-this- # file lines came from a LOCATION lookup — nothing was matched, so there # is no matching passage and the body's opening would be a fabricated # reason. Absence here is meaningful: a line with no passage under it is # one that earned its place by where it lives, not by what it says. # WHOLE, and only on first sight: a `seen` line is a pointer to # a record already in context, and its passage is already there too. if item.get("seen"): continue passage = _menu_passage( item.get("doc_title") or item.get("title"), (wp_chunks.get(int(item["id"])) or {}).get("text"), item.get("name") or "", ) if passage: lines.append(f"> ↳ {passage}") if suggested: lines.append(_suggest_line(path, suggested)) 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, yet it fires on # the strongest possible claim ("there is already a canonical helper in # this exact file") and must be measurable. The sync class gets its own # tag: its pull-through rate is the number that says whether edit-time # record-sync actually happens. by_arm: dict[str, list[int]] = {} if sync_note_ids: by_arm["write_path_sync"] = list(sync_note_ids) for marker, item in menu: # A rendered repeat is not a new surfacing — same cut the log row # takes, so this table and `retrieval_logs` keep agreeing about the # same call. The semantic arm's rows (`write_path_semantic`) # were written by the pipeline beside its call row, so only the # lookups are booked here. if int(item["id"]) in excluded or not marker.startswith("nearby"): continue by_arm.setdefault("write_path_place", []).append(int(item["id"])) for arm, ids in by_arm.items(): record_surfaced( user_id=user_id, note_ids=ids, source=arm, project_id=project_id, ) # ── Standing rules that may apply here ─────────────────────────────── # # A SUGGESTION, not a binding surface, and the distinction is the design: # 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 RULE IS ELIGIBLE, including one already named this session: a # rule the session was told about an hour ago is not a rule in front of # the reader now. # # The arm itself is `retrieval_pipeline.run_rule_arm`: the band, the # fresh/repeat split, the unconditional call row and the fresh-only # surfacing rows are written there once for all three # rule arms. What is decided HERE is only what makes this moment this # moment — the query is the code being written (or the path, for an edit # with no body), and a line says the rule may apply "here". # # Fails open like every other arm: a rule hint must never break a write. # The direct band, then a rule reached through a linked lesson, # composed by `_rule_moment` exactly as the other two rule builders are. rule_ids: list[int] = [] shown_rule_ids: list[int] = [] checkpoint: dict = {} try: result = await _rule_moment( rp.WRITE_PATH_RULE, rp.RuleMoment( user_id=user_id, query=code or path, project_id=project_id, where="here", checkpoint_where="here", exclude=frozenset(exclude_rule_ids or []), held=frozenset(held_rule_ids or []), ), floor=cfg["rule_threshold"], budget=cfg["rule_top_k"], checkpoint_floor=cfg["checkpoint_threshold"], ) lines.extend(result.lines) rule_ids.extend(result.rule_ids) # Every rule line the band showed, repeats included — what the # soft-link recorder pairs with this response's lessons. shown_rule_ids = result.shown_rule_ids checkpoint = result.checkpoint except Exception: logger.debug("write-path rule arm failed", exc_info=True) # A lesson and a rule on this one response: the soft-link # recorder counts the pair, keyed on the FILE — every edit to one file is # one situation — and asks about it once the evidence holds. Fails open. shown_lessons = [ int(item["id"]) for _m, item in menu if item.get("kind") == LESSON_NOTE_TYPE ] proposal = await lesson_rules_svc.co_surfaced( user_id, shown_lessons, shown_rule_ids, arm="w", situation=path, project_id=project_id or None, ) if proposal: lines.append(proposal) return { "context": "\n".join(lines), "note_ids": note_ids, "sync_note_ids": sync_note_ids, "config": cfg, "suggested": suggested, "divergence": divergence, "derive": derive, "derive_keys": [d["key"] for d in derive] + ( [design_dedup] if design_dedup else [] ), "rule_ids": rule_ids, "checkpoint": checkpoint, "ruling_system_ids": rulings["system_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, root: str = "", cwd: str = "", seen_ruling_systems: list[int] | None = None, ) -> dict: """Standing rules that may apply to the ACTION about to be taken — matched directly, then reached through a linked lesson (`_rule_moment`) — and the rulings of any area whose files the command names. The sibling of the write-path rule arm. That arm is keyed on `code or path`, so it retrieves only 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 — arrives here instead, at the moment it matters. WHY A MECHANICAL TRIGGER AND NOT AN INSTRUCTION. 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 — present in context and salient at the moment are different properties, and only the second is what this arm is for. `root` and `cwd` are the repo's absolute root and the command's working directory, from the hook; they are what turn the paths a command names into the repo-relative paths a System's patterns are written in. Fails open and returns an empty context on any error: a recall aid may never break the operator's action. """ # `checkpoint` is present on EVERY return, including the early ones. The # two arms feed one shell reader, and a key that exists on some responses # and not others is read there as an empty variable either way — so the # difference is invisible at the point it would bite and only shows up in # a test that asserts the contract. Same reason `warnings` is always a # list in the telemetry readout: an absent key and an empty one must not # be two ways of saying nothing. out: dict = {"context": "", "rule_ids": [], "checkpoint": {}} 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. result = await _rule_moment( rp.PRE_TOOL_RULE, rp.RuleMoment( user_id=user_id, query=command[:_TOOL_QUERY_CHARS], project_id=project_id, where=f"to this {tool_name} call", checkpoint_where=f"this {tool_name} call", exclude=frozenset(exclude_rule_ids or []), held=frozenset(held_rule_ids or []), ), floor=cfg["tool_rule_threshold"], budget=cfg["tool_rule_top_k"], checkpoint_floor=cfg["checkpoint_threshold"], ) if result.lines: out["context"] = "\n".join(result.lines) out["rule_ids"] = result.rule_ids out["checkpoint"] = result.checkpoint except Exception: logger.debug("pre-tool rule arm failed", exc_info=True) # Its own arm, not part of the rule search: a lookup by path, with no # floor and no budget, so it adds to the rule lines rather than competing # with them. First, because a ruling is the operator's own decision. # `ruling_system_ids` is set only when a ruling was shown: this arm fires # on every command, and the hook reads an absent list as an empty one. try: paths = system_rulings_svc.command_paths(command, root=root, cwd=cwd) if project_id else [] if paths and (await get_writepath_config(user_id)).get("enabled"): rulings = await system_rulings_svc.rulings_for_paths( user_id, project_id, paths, seen=seen_ruling_systems, source="rulings_pre_tool", ) if rulings["lines"]: out["context"] = "\n".join( rulings["lines"] + ([out["context"]] if out.get("context") else []) ) out["ruling_system_ids"] = rulings["system_ids"] except Exception: logger.debug("pre-tool rulings arm failed", exc_info=True) return out # --- the design-guidance arm ------------------------------------------------ # A design system BINDS like a rule. The session-start block names it and the # call that reads its prose, which is complete for a session that knows to ask # and silent for one that is writing a component — the gap every unasked arm # exists to close. # # A TRIGGER, NOT A SEARCH. A project has exactly one design system # (projects.design_system_id is a single FK), so there is nothing to rank and # no vector to compute: the question "does this guidance apply here" is # answered by the file being UI. Deterministic and cheap, and it takes no # slot from the ranked menu — the band, floor and budget the other arms were # tuned against are untouched by construction, which is why this adds no # retrieval_logs row: there is no score distribution for it to join. # # AN INDEX, NOT THE PROSE. Resolved guidance runs to thousands of characters # (a house style is long by nature), which would take most of the hook's # additionalContext cap on its own. So the line names the SECTIONS of each # inherited layer — the headings are self-describing ("Where the accent must # NOT appear", "Voice and tone") the way rule titles are — and inlines only a # layer short enough to be a line: in practice the leaf, since a child system # holds just its departure from the house style. Choosing a paragraph by # meaning would need the guidance embedded per section; that is justified # only if this index turns out not to be read. # # ONCE PER SESSION PER SYSTEM, on the hook's token-keyed channel # (`exclude_derive`, keyed `design:`). That channel already dedups opaque # keys on its own file, so the arm needs no new plugin state. _DESIGN_UI_EXTENSIONS = frozenset({ ".vue", ".svelte", ".css", ".scss", ".sass", ".less", ".tsx", ".jsx", ".html", }) # A guidance layer this short is shown whole; anything longer is indexed. _DESIGN_INLINE_CHARS = 500 _DESIGN_HEADING = re.compile(r"^##\s+(.+?)\s*$", re.M) def design_key(design_system_id: int) -> str: """The dedup token for the design arm on the hook's keyed channel.""" return f"design:{int(design_system_id)}" def is_ui_path(path: str) -> bool: """Whether writing `path` is writing UI — the design arm's trigger.""" name = (path or "").rsplit("/", 1)[-1].lower() return any(name.endswith(ext) for ext in _DESIGN_UI_EXTENSIONS) def _design_line(path: str, design: dict) -> str: """Name the design system that binds this file, and what its prose covers.""" ds_id = design["id"] layers: list[str] = [] for layer in design.get("guidance") or []: text = (layer.get("guidance") or "").strip() if not text: continue flat = " ".join(text.split()) headings = _DESIGN_HEADING.findall(text) if len(flat) <= _DESIGN_INLINE_CHARS: layers.append(f"{layer['title']}: \"{flat}\"") elif headings: layers.append(f"{layer['title']} covers " + " · ".join(headings)) else: short, _cut = elide(flat, _DESIGN_INLINE_CHARS) layers.append(f"{layer['title']}: \"{short}\"") inherits = ( " (inherits " + " › ".join(design["inherits_from"]) + ")" if design.get("inherits_from") else "" ) out = ( f"> Design system binds `{path}`: {design['title']} (id {ds_id}){inherits}. " f"Read `get_design_system({ds_id})` → `resolved_guidance` before writing " f"UI here, and take values from `resolve_design_system({ds_id})` rather " f"than hand-writing them." ) if layers: out += " " + "; ".join(layers) + "." return out + " (Shown once per session.)" async def _design_arm( user_id: int, project_id: int, path: str, skip: set[str], ) -> tuple[str, str]: """(line, dedup key) for a UI write in a project with a design system, or ("", "") — never raises: a design hint must never break a write.""" if not project_id or not is_ui_path(path): return "", "" try: project = await projects_svc.get_project(user_id, project_id) ds_id = getattr(project, "design_system_id", None) if project else None if not ds_id or design_key(ds_id) in skip: return "", "" design = await design_systems_svc.design_context(user_id, ds_id) if not design: return "", "" return _design_line(path, design), design_key(ds_id) except Exception: logger.debug("write-path design arm failed", exc_info=True) return "", "" def _derive_line(path: str, derive: list[dict]) -> str: """The ledger's word on the names being written: 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 — 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: 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.""" 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 _suggest_line(path: str, suggested: list[dict]) -> str: """One line naming the recorded shape this code looks like, as evidence for the judgment the agent gives at the end of the turn — not a record of one already made.""" by_snippet: dict[int, list[str]] = {} for row in suggested: 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)} → looks like #{sid}" for sid, names in by_snippet.items() ] return ( f"> Shape accounting: at `{path}` — {'; '.join(parts)} " "(you pulled that snippet this session and this code names or resembles " "it). When you judge this turn's shapes, say whether it is: `instance` " "of it, `variant` with the why, or something else." ) def _goal_line(goal: str, project_id: int) -> str: """The Goal line, trimmed at a word break with a visible cut. A raw slice would end mid-word with nothing to say more exists, and a reader would take half a sentence for the whole goal. """ 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 = "", source: str = "", session_id: 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 `.grimoire` marker file naming the project directly — the only key a session outside a git repo has. 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. source / session_id: the host's SessionStart `source` and the session's id, when the adapter sends them. They decide what the claim section says (`task_claims.render_claims`): a compaction gets back the work it had claimed with its latest logs; a new session hears about other sessions' live and abandoned claims; a resume hears nothing. Returns {"context": str, "project": dict | None}. `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. It 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 Grimoire is the skill's to say, and it says it once. """ lines: list[str] = ["# Grimoire — 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 without this # the standards would reach only 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`.", ] # The claim section goes after the project block and before any "nothing # loaded" note: claimed work is the most specific thing this session can be # told, and it is true whether or not a project resolved. Best-effort — a # session start never fails on it. try: lines += await task_claims_svc.claims_for_session_start( user_id, project_dict["id"] if project_dict else 0, source, session_id, ) except Exception: # noqa: BLE001 - context is best-effort logger.warning("claim section skipped", exc_info=True) # Nothing loaded — say which nothing. An id that was SENT and did not # resolve is the 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 Grimoire " "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 Grimoire project, so " "no project context was loaded. Bind it once with " f'`bind_repo(repo_url="{unbound_repo}", project_id=)` ' "(call `list_projects` to find the id) and future sessions here will " "auto-load that project's context.", ] else: lines += ["", "No Grimoire 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, }