feat(learning): prospective held-out style-distance trend (Path A)

A clean held-out test of voice learning was no longer runnable: every
final-uploaded case already has its lessons folded, and lessons are stored
universal/untagged so leave-one-out is impossible. Path A makes the test
prospective instead — capture the generalization datapoint at the one moment
it's clean.

On final upload, after the draft↔final pair is created but BEFORE this
case's lessons are folded (folding is a separate manual /training step), we
snapshot style_distance (anti_pattern_total, golden-ratio max-deviation,
change_percent) alongside the current voice-lesson pool size. Because the
draft was written with only the PRIOR pool, each row is a clean "with N
accumulated lessons, our draft on this unseen case scored X" datapoint. As
the pool grows over cases, a downward trend = learning generalizes.

- db: SCHEMA_V45 style_distance_history (append-only) + helpers
  voice_lesson_pool_sizes / record_style_distance_snapshot /
  get_style_distance_history.
- app: best-effort capture in api_upload_final_decision (never fails the
  upload); GET /api/learning/style-distance-history for the trend.

Reuses the existing style_distance service + appeal_type_rules pool — no
parallel metric path. The 8 existing cases are already folded, so the table
starts empty and fills from the next final (their clean window is past).

Invariants: G2 (reuse style_distance/appeal_type_rules — one path),
INV-LRN4 (measure the draft↔final gap; this is its trend surface). LLM-free
(style_distance is deterministic) so it runs in the container.

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

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

@@ -1748,6 +1748,29 @@ CREATE INDEX IF NOT EXISTS idx_case_precedents_argument ON case_precedents(argum
 """
 
 
+# V45 (Path A — prospective held-out): a style-distance snapshot captured at
+# final-upload time, BEFORE this case's lessons are folded. Because the draft was
+# written with only the PRIOR lesson pool, each row is a clean generalization
+# datapoint: "with N accumulated lessons, our draft on this unseen case scored X".
+# pool_* record how much voice-learning had accumulated when the draft was made,
+# so the trend (style-distance vs pool size / time) shows whether learning
+# generalizes. Append-only; one row per final upload. (07-learning §0, INV-LRN4.)
+SCHEMA_V45_SQL = """
+CREATE TABLE IF NOT EXISTS style_distance_history (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    case_number TEXT NOT NULL,
+    pair_id UUID REFERENCES draft_final_pairs(id) ON DELETE SET NULL,
+    measured_at TIMESTAMPTZ DEFAULT now(),
+    pool_discussion_rules INT DEFAULT 0,
+    pool_transition_phrases INT DEFAULT 0,
+    anti_pattern_total INT,
+    ratio_max_deviation REAL,
+    change_percent REAL
+);
+CREATE INDEX IF NOT EXISTS idx_style_distance_history_measured ON style_distance_history(measured_at);
+"""
+
+
 # Stable, arbitrary key for the session-level advisory lock that serialises
 # schema DDL across processes. Every short-lived process (cron drains, services)
 # re-runs the idempotent migrations on startup; without this lock two processes
@@ -1814,6 +1837,7 @@ async def _apply_schema_ddl(conn: asyncpg.Connection) -> None:
         await conn.execute(SCHEMA_V42_SQL)
         await conn.execute(SCHEMA_V43_SQL)
         await conn.execute(SCHEMA_V44_SQL)
+        await conn.execute(SCHEMA_V45_SQL)
 
 
 async def init_schema() -> None:
@@ -2859,6 +2883,66 @@ async def get_methodology_overrides(category: str) -> dict:
     return out
 
 
+async def voice_lesson_pool_sizes() -> dict:
+    """How many folded voice-learning items are in the writer-consumed pool right now
+    (universal discussion_rules + transition_phrases). Used by the prospective held-out
+    snapshot to record how much learning had accumulated when a draft was produced."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            "SELECT rule_category, COALESCE(jsonb_array_length(rule_value), 0) AS n "
+            "FROM appeal_type_rules "
+            "WHERE appeal_type = '_global' AND rule_key = 'universal' "
+            "AND rule_category IN ('discussion_rules', 'transition_phrases')",
+        )
+    out = {"discussion_rules": 0, "transition_phrases": 0}
+    for r in rows:
+        out[r["rule_category"]] = r["n"]
+    return out
+
+
+async def record_style_distance_snapshot(
+    case_number: str, pair_id: str | None, pool_rules: int, pool_phrases: int,
+    anti_pattern_total: int | None, ratio_max_deviation: float | None,
+    change_percent: float | None,
+) -> None:
+    """Append one prospective held-out datapoint (Path A). Append-only; never updates."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(
+            "INSERT INTO style_distance_history (case_number, pair_id, "
+            "pool_discussion_rules, pool_transition_phrases, anti_pattern_total, "
+            "ratio_max_deviation, change_percent) VALUES ($1, $2, $3, $4, $5, $6, $7)",
+            case_number, UUID(pair_id) if pair_id else None,
+            pool_rules, pool_phrases, anti_pattern_total,
+            ratio_max_deviation, change_percent,
+        )
+
+
+async def get_style_distance_history() -> list[dict]:
+    """The prospective held-out trend, oldest-first (Path A). Each row = one final
+    upload's style-distance measured before that case's lessons were folded."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            "SELECT case_number, measured_at, pool_discussion_rules, "
+            "pool_transition_phrases, anti_pattern_total, ratio_max_deviation, "
+            "change_percent FROM style_distance_history ORDER BY measured_at",
+        )
+    return [
+        {
+            "case_number": r["case_number"],
+            "measured_at": r["measured_at"].isoformat() if r["measured_at"] else None,
+            "pool_discussion_rules": r["pool_discussion_rules"],
+            "pool_transition_phrases": r["pool_transition_phrases"],
+            "anti_pattern_total": r["anti_pattern_total"],
+            "ratio_max_deviation": r["ratio_max_deviation"],
+            "change_percent": r["change_percent"],
+        }
+        for r in rows
+    ]
+
+
 async def get_recent_decision_lessons(limit: int = 15, practice_area: str = "") -> list[dict]:
     """Per-decision learnings the chair/curator attached in /training (decision_lessons),
     so the writer consumes them too (T15). Prefers style/structure/lexicon, recent first.

web/app.py

@@ -3813,6 +3813,25 @@ async def api_upload_final_decision(case_number: str, file: UploadFile = File(..
     # checking + missing-precedent flagging (07-learning §1.3). Surfaced, not silent.
     library = await _enroll_final_in_library(case, case_number, final_text, chair_name)
 
+    # Path A — prospective held-out snapshot. Right now the draft (decision_blocks)
+    # was written with only the PRIOR lesson pool — this case's lessons fold later,
+    # manually, in /training. So measuring style-distance HERE is a clean
+    # generalization datapoint. Append-only; best-effort (never fails the upload).
+    try:
+        from legal_mcp.services import style_distance as _sd
+        sd = await _sd.style_distance(case_number)
+        if isinstance(sd, dict) and "error" not in sd:
+            pool = await db.voice_lesson_pool_sizes()
+            summ = sd.get("summary", {})
+            await db.record_style_distance_snapshot(
+                case_number, pair_id,
+                pool.get("discussion_rules", 0), pool.get("transition_phrases", 0),
+                summ.get("anti_pattern_total"), summ.get("ratio_max_deviation_pp"),
+                summ.get("change_percent"),
+            )
+    except Exception as e:
+        logger.warning("held-out style-distance snapshot failed for %s: %s", case_number, e)
+
     case_dir = config.find_case_dir(case_number)
     if case_dir.exists():
         commit_and_push(case_dir, f"החלטה סופית של היו\"ר: {final_name}")
@@ -4738,6 +4757,15 @@ async def api_learning_style_distance(case_number: str):
     return await _sd.style_distance(case_number)
 
 
+@app.get("/api/learning/style-distance-history")
+async def api_learning_style_distance_history():
+    """Path A — מגמת ה-held-out הפרוספקטיבי: snapshot של מרחק-הסגנון שנלכד בכל
+    העלאת-סופי *לפני* הטמעת-לקחי-התיק, מול גודל-בריכת-הלקחים באותו רגע. ירידה
+    ב-anti_pattern_total / change_percent ככל שהבריכה גדלה = הלמידה מכלילה."""
+    items = await db.get_style_distance_history()
+    return {"items": items, "count": len(items)}
+
+
 def _coerce_json(raw):
     if isinstance(raw, str):
         try: