Merge pull request 'feat(learning): prospective held-out style-distance trend (Path A)' (#337) from worktree-style-distance-history into main

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.