feat(learning): grow style-exemplars from every final (P1 #5)

The style-exemplar corpus (channel B — the writer's block-level retrieval of
Dafna's real prose) was FROZEN at the one-time seed backfill: new finals were
enrolled into style_corpus but never broken into exemplars, so the richest style
channel never grew (8137/8126/8174 had 0 exemplars). The writer kept retrieving
only March–April seed paragraphs no matter how many finals were signed.

Extract the per-decision exemplar logic (section→paragraph→Voyage-embed→replace)
into a shared service `legal_mcp.services.style_exemplars.extract_and_store` —
the SINGLE implementation now used by BOTH the one-time backfill and the live
enrollment path (G2; no parallel extractor). `_enroll_final_in_library` calls it
on every final upload (source='internal_committee', the same source the writer's
search_style_exemplars reads). Voyage embeds over REST → container-safe;
best-effort, surfaced in the upload response, never fails the upload.

Effect: every signed final now grows the exemplar corpus, so the writer's
block-level style retrieval improves with each decision — the core "learn from
every decision" fix for channel B. Path A (style_distance_history) will track
whether the larger exemplar pool reduces style-distance over time.

Invariants: G2 (one extraction path shared by backfill + enroll), INV-LRN5
(style/structure prose only — substance routes elsewhere), INV-LRN4 (the
draft↔final loop now feeds the exemplar channel, not just the lesson channel).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mcp-server/src/legal_mcp/services/style_exemplars.py

@@ -0,0 +1,83 @@
+"""Block-level style-exemplar extraction (channel B of Style Acquisition).
+
+Splits one of Dafna's decisions into section→paragraph units, embeds them
+(Voyage), and stores them in `style_exemplars` so the writer can retrieve real
+block-level prose by section/outcome/practice_area (07-learning §0.2 channel B).
+
+This is the SINGLE source of truth for exemplar extraction (G2): both the
+one-time backfill (`scripts/backfill_style_exemplars.py`) and the live
+final-enrollment path (`_enroll_final_in_library`) call `extract_and_store`.
+Before this, exemplars were frozen at the seed backfill — new finals enrolled
+into style_corpus but were never broken into exemplars, so the richest style
+channel never grew. Embedding is Voyage-over-REST → container-safe.
+"""
+from __future__ import annotations
+
+import logging
+
+from legal_mcp.services import db, embeddings
+from legal_mcp.services.chunker import _split_into_sections
+
+logger = logging.getLogger(__name__)
+
+# chunker section_type → style_exemplars.section
+_SECTION_MAP = {
+    "facts": "background",
+    "appellant_claims": "claims",
+    "respondent_claims": "claims",
+    "legal_analysis": "discussion",
+    "conclusion": "summary",
+    "ruling": "summary",
+    "intro": "other",
+    "other": "other",
+}
+
+MIN_WORDS = 25      # skip tiny fragments
+MAX_WORDS = 450     # skip over-long blobs (likely un-split)
+MAX_PER_SECTION = 15
+
+
+def _paragraphs(section_text: str) -> list[str]:
+    """Split a section into paragraph units (blank-line separated; fall back to lines)."""
+    raw = [p.strip() for p in section_text.split("\n\n")]
+    if len(raw) <= 1:
+        raw = [p.strip() for p in section_text.split("\n")]
+    out = []
+    for p in raw:
+        wc = len(p.split())
+        if MIN_WORDS <= wc <= MAX_WORDS:
+            out.append(p)
+    return out[:MAX_PER_SECTION]
+
+
+def units_for(full_text: str) -> list[tuple[str, str]]:
+    """(section, paragraph) units for a decision — pure, no I/O."""
+    units: list[tuple[str, str]] = []
+    for section_type, section_text in _split_into_sections(full_text or ""):
+        section = _SECTION_MAP.get(section_type, "other")
+        for para in _paragraphs(section_text):
+            units.append((section, para))
+    return units
+
+
+async def extract_and_store(
+    decision_number: str, source: str, full_text: str,
+    practice_area: str = "", outcome: str = "",
+) -> int:
+    """Idempotently (re)build a decision's block-level style exemplars: split →
+    embed → replace. Returns the number of exemplars stored. Raises on failure;
+    callers in best-effort paths (enroll) should wrap in try/except."""
+    units = units_for(full_text)
+    if not units:
+        return 0
+    texts = [u[1] for u in units]
+    vecs = await embeddings.embed_texts(texts, input_type="document")
+    await db.delete_style_exemplars(decision_number, source)
+    for (section, para), vec in zip(units, vecs):
+        await db.insert_style_exemplar(
+            decision_number=decision_number, source=source,
+            practice_area=practice_area, outcome=outcome,
+            section=section, paragraph_text=para, word_count=len(para.split()),
+            embedding=vec,
+        )
+    return len(units)

scripts/backfill_style_exemplars.py

@@ -19,40 +19,15 @@ import argparse
 import asyncio
 import logging
 
-from legal_mcp.services import db, embeddings
+from legal_mcp.services import db
-from legal_mcp.services.chunker import _split_into_sections
+from legal_mcp.services import style_exemplars as sx
 
 logging.basicConfig(level=logging.INFO, format="%(message)s")
 log = logging.getLogger("backfill_exemplars")
 
-# chunker section_type → style_exemplars.section
+# Section mapping + paragraph splitting now live in the shared service
-_SECTION_MAP = {
+# (legal_mcp.services.style_exemplars) so the backfill and the live
-    "facts": "background",
+# final-enrollment path use ONE extraction implementation (G2).
-    "appellant_claims": "claims",
-    "respondent_claims": "claims",
-    "legal_analysis": "discussion",
-    "conclusion": "summary",
-    "ruling": "summary",
-    "intro": "other",
-    "other": "other",
-}
-
-MIN_WORDS = 25      # skip tiny fragments
-MAX_WORDS = 450     # skip over-long blobs (likely un-split)
-MAX_PER_SECTION = 15
-
-
-def _paragraphs(section_text: str) -> list[str]:
-    """Split a section into paragraph units (blank-line separated; fall back to lines)."""
-    raw = [p.strip() for p in section_text.split("\n\n")]
-    if len(raw) <= 1:
-        raw = [p.strip() for p in section_text.split("\n")]
-    out = []
-    for p in raw:
-        wc = len(p.split())
-        if MIN_WORDS <= wc <= MAX_WORDS:
-            out.append(p)
-    return out[:MAX_PER_SECTION]
 
 
 async def _gather_sources() -> list[dict]:
@@ -94,28 +69,18 @@ async def main(apply: bool) -> None:
 
     total_paras = 0
     for src in sources:
-        units: list[tuple[str, str]] = []  # (section, paragraph)
+        n = len(sx.units_for(src["full_text"]))
-        for section_type, section_text in _split_into_sections(src["full_text"]):
+        if not n:
-            section = _SECTION_MAP.get(section_type, "other")
-            for para in _paragraphs(section_text):
-                units.append((section, para))
-        if not units:
             continue
-        total_paras += len(units)
+        total_paras += n
-        log.info("  %-14s %-16s → %d פסקאות", src["source"], src["decision_number"], len(units))
+        log.info("  %-14s %-16s → %d פסקאות", src["source"], src["decision_number"], n)
         if not apply:
             continue
-
+        await sx.extract_and_store(
-        await db.delete_style_exemplars(src["decision_number"], src["source"])
+            decision_number=src["decision_number"], source=src["source"],
-        texts = [u[1] for u in units]
+            full_text=src["full_text"], practice_area=src["practice_area"],
-        vecs = await embeddings.embed_texts(texts, input_type="document")
+            outcome=src["outcome"],
-        for (section, para), vec in zip(units, vecs):
+        )
-            await db.insert_style_exemplar(
-                decision_number=src["decision_number"], source=src["source"],
-                practice_area=src["practice_area"], outcome=src["outcome"],
-                section=section, paragraph_text=para, word_count=len(para.split()),
-                embedding=vec,
-            )
 
     if apply:
         cov = await db.count_style_exemplars()

web/app.py

@@ -3769,6 +3769,22 @@ async def _enroll_final_in_library(
         logger.warning("inline metadata extraction failed for %s: %s", case_number, e)
         out["metadata_error"] = str(e)
 
+    # Grow the style-exemplar corpus (channel B): break this final into block-level
+    # paragraphs the writer retrieves (07-learning §0.2). Before this, exemplars were
+    # frozen at the one-time seed backfill — new finals never got exemplarized, so the
+    # richest style channel never grew. Same extraction the backfill uses (G2). Voyage
+    # embeds over REST → container-safe. Best-effort; surfaced, never fails the upload.
+    try:
+        from legal_mcp.services import style_exemplars as _sx
+        n_ex = await _sx.extract_and_store(
+            decision_number=case_number, source="internal_committee",
+            full_text=final_text, practice_area=case.get("practice_area", ""),
+        )
+        out["exemplars"] = n_ex
+    except Exception as e:
+        logger.warning("style-exemplar extraction failed for %s: %s", case_number, e)
+        out["exemplars_error"] = str(e)
+
     # The precedents this decision cites → link to the library; flag the ones not found.
     try:
         await cit_tools.extract_internal_citations(case_law_id=case_law_id, limit=0)