feat(plugin): knowledge auto-inject (Path A) — title-first per-turn awareness

New UserPromptSubmit hook (scribe_autoinject.sh) + GET /api/plugin/retrieve that
surface the TITLES (never bodies) of the few notes clearing four anti-bloat
gates: a per-user confidence threshold (stricter than pull search), a margin
gate, per-session dedup (exclude_ids), and a top-k ceiling. Each retrieval is
logged to retrieval_logs as source=auto_inject so the threshold can be tuned
from data. Per-user config (enable / threshold / top-k) is DB-backed via
/api/settings with a Settings UI card; defaults enabled, threshold 0.55,
top-k 3 (conservative — tune once auto_inject telemetry accrues).

Scribe: project 2, milestone 93, task 1033.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Xz4j1H7pjYSjKsEpgcNH5E

src/scribe/routes/plugin.py

@@ -57,6 +57,48 @@ async def session_context():
     return jsonify(result)
 
 
+@plugin_bp.get("/retrieve")
+@login_required
+async def autoinject_retrieve():
+    """Title-first knowledge auto-inject for the plugin's UserPromptSubmit hook.
+
+    Given the user's prompt (`q`), returns a compact awareness hint — note
+    titles + scores only, never bodies — for the top hits that clear the
+    per-user gates (see services.plugin_context.build_autoinject_hint). Returns
+    empty context (most of the time) when disabled or nothing is relevant.
+
+    Query:
+        q (str)            — the user's prompt to retrieve against.
+        repo (optional)    — working repo remote; resolved to the bound project
+                             to scope the search (mirrors /context). Unbound or
+                             absent → searches all the user's notes.
+        project_id (opt)   — explicit project scope override (ad-hoc/testing).
+        exclude_ids (opt)  — comma-separated note ids already injected this
+                             session; skipped so each note injects at most once.
+    """
+    q = (request.args.get("q") or "").strip()
+    try:
+        project_id = int(request.args.get("project_id", 0) or 0)
+    except (TypeError, ValueError):
+        project_id = 0
+
+    repo = (request.args.get("repo") or "").strip()
+    if repo and not project_id:
+        resolved = await repo_bindings_svc.resolve_project(g.user.id, repo)
+        if resolved:
+            project_id = resolved
+
+    exclude_ids = [
+        int(p) for p in (request.args.get("exclude_ids") or "").split(",")
+        if p.strip().isdigit()
+    ]
+
+    result = await plugin_ctx_svc.build_autoinject_hint(
+        g.user.id, q, project_id=project_id, exclude_ids=exclude_ids
+    )
+    return jsonify(result)
+
+
 @plugin_bp.get("/processes")
 @login_required
 async def process_manifest():

src/scribe/services/plugin_context.py

@@ -15,6 +15,7 @@ index alone already steers behavior.
 from __future__ import annotations
 
 import re
+import time
 
 from sqlalchemy import select
 
@@ -24,6 +25,9 @@ from scribe.services import knowledge as knowledge_svc
 from scribe.services import notes as notes_svc
 from scribe.services import projects as projects_svc
 from scribe.services import rulebooks as rulebooks_svc
+from scribe.services.embeddings import semantic_search_notes
+from scribe.services.retrieval_telemetry import record_retrieval
+from scribe.services.settings import get_setting
 
 # Defensive cap below Claude Code's 10k additionalContext limit.
 _MAX_CHARS = 9000
@@ -31,6 +35,28 @@ _MAX_CHARS = 9000
 # Max chars of a Process body to fold into the auto-surface description.
 _PROC_PREVIEW_CHARS = 200
 
+# --- Knowledge auto-inject (Path A: per-turn awareness push) -----------------
+# Per-user settings (keys live in the generic settings table). The threshold is
+# deliberately STRICTER than the pull-search default (embeddings
+# DEFAULT_SIMILARITY_THRESHOLD = 0.45): an unsolicited per-turn inject must clear
+# a higher bar than a search the agent chose to run. Defaults start conservative
+# and are meant to be tuned from retrieval_logs (source='auto_inject') once data
+# accrues — they're exposed in the Settings UI, no restart needed.
+AUTOINJECT_ENABLED_KEY = "kb_autoinject_enabled"
+AUTOINJECT_THRESHOLD_KEY = "kb_autoinject_threshold"
+AUTOINJECT_TOP_K_KEY = "kb_autoinject_top_k"
+
+AUTOINJECT_DEFAULT_ENABLED = True
+AUTOINJECT_DEFAULT_THRESHOLD = 0.55
+AUTOINJECT_DEFAULT_TOP_K = 3
+
+# Margin gate: drop any hit more than this far below the top hit's score, so a
+# single strong match doesn't drag in a wall of barely-passing neighbours.
+_AUTOINJECT_BAND = 0.10
+# Hard ceiling on top-k regardless of the user's setting — this is an
+# awareness menu (titles only), never a content dump.
+_AUTOINJECT_MAX_TOP_K = 10
+
 
 def _slugify(text: str) -> str:
     """kebab-case slug for a skill directory name (a-z0-9 + single hyphens)."""
@@ -82,6 +108,95 @@ async def build_process_manifest(user_id: int) -> dict:
     return {"processes": procs, "total": len(procs)}
 
 
+async def get_autoinject_config(user_id: int) -> dict:
+    """Resolve a user's auto-inject settings, falling back to the defaults.
+
+    Returns {"enabled": bool, "threshold": float, "top_k": int}, clamped to
+    sane ranges (threshold to [0,1]; top_k to [1, _AUTOINJECT_MAX_TOP_K]).
+    """
+    enabled_raw = await get_setting(
+        user_id, AUTOINJECT_ENABLED_KEY,
+        "true" if AUTOINJECT_DEFAULT_ENABLED else "false",
+    )
+    enabled = enabled_raw.strip().lower() in ("true", "1", "yes", "on")
+
+    try:
+        threshold = float(await get_setting(
+            user_id, AUTOINJECT_THRESHOLD_KEY, str(AUTOINJECT_DEFAULT_THRESHOLD)))
+    except (TypeError, ValueError):
+        threshold = AUTOINJECT_DEFAULT_THRESHOLD
+    threshold = min(1.0, max(0.0, threshold))
+
+    try:
+        top_k = int(float(await get_setting(
+            user_id, AUTOINJECT_TOP_K_KEY, str(AUTOINJECT_DEFAULT_TOP_K))))
+    except (TypeError, ValueError):
+        top_k = AUTOINJECT_DEFAULT_TOP_K
+    top_k = min(_AUTOINJECT_MAX_TOP_K, max(1, top_k))
+
+    return {"enabled": enabled, "threshold": threshold, "top_k": top_k}
+
+
+async def build_autoinject_hint(
+    user_id: int,
+    query: str,
+    project_id: int = 0,
+    exclude_ids: list[int] | None = None,
+) -> dict:
+    """Title-first awareness hint for the plugin's UserPromptSubmit hook.
+
+    The four anti-bloat gates (see the module + milestone-93 design):
+      1. high-confidence threshold (stricter than pull) — set per-user;
+      2. margin gate — keep only hits within _AUTOINJECT_BAND of the top score;
+      3. session dedup — caller passes already-injected ids as `exclude_ids`;
+      4. title-first payload — id + title + score only, never bodies.
+    Disabled, blank-query, or nothing-clears-the-gates all return empty context,
+    so most turns inject nothing.
+
+    Returns {"context": str, "note_ids": list[int], "config": dict}. Every
+    retrieval (even empty) is logged to retrieval_logs as source='auto_inject'
+    so the threshold can be tuned from data.
+    """
+    cfg = await get_autoinject_config(user_id)
+    empty = {"context": "", "note_ids": [], "config": cfg}
+    q = (query or "").strip()
+    if not cfg["enabled"] or not q:
+        return empty
+
+    t0 = time.perf_counter()
+    hits = await semantic_search_notes(
+        user_id, q,
+        limit=cfg["top_k"],
+        threshold=cfg["threshold"],
+        project_id=(project_id or None),
+        exclude_ids=set(exclude_ids or []),
+    )
+    record_retrieval(
+        user_id=user_id, source="auto_inject", query=q,
+        threshold=cfg["threshold"], limit=cfg["top_k"],
+        project_id=(project_id or None), is_task=None, results=hits,
+        duration_ms=(time.perf_counter() - t0) * 1000.0,
+    )
+    if not hits:
+        return empty
+
+    # Margin gate: keep only hits close to the strongest one.
+    top_score = hits[0][0]
+    kept = [(s, n) for s, n in hits if s >= top_score - _AUTOINJECT_BAND]
+
+    lines = [
+        "> Possibly relevant from your Scribe notes — call `get_note(id)` to "
+        "open any in full (titles only; injected once per session):",
+    ]
+    note_ids: list[int] = []
+    for score, note in kept:
+        note_ids.append(int(note.id))
+        title = (note.title or "(untitled)").replace("\n", " ").strip()
+        lines.append(f"> - #{note.id} \"{title}\" ({score:.2f})")
+
+    return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg}
+
+
 async def _topic_titles(topic_ids: set[int]) -> dict[int, str]:
     """Map topic_id -> title for the given ids (live topics only)."""
     if not topic_ids: