Add SMALL-model domain classifier for Penyusun

Replaces the keyword-only domain anchor with a learned classifier that
extracts (sektor, jenjang, instansi, kasus_terkait, kerangka_hukum,
anti_drift) from topic + research + extra_instructions. The keyword path
remains as a fallback for when the SMALL model produces unparseable JSON.

Composition in _derive_domain_constraints:
1. _CASE_FACTS keyword overrides — always applied for known cases
(e.g. Kasus Ibam) since hand-curated facts beat generated ones.
2. classify_domain() SMALL-model call — produces DomainAnalysis,
rendered as constraint string into the BIG system prompt.
3. _keyword_domain_constraints fallback — fires only when classifier
raises (parse error, transport error, validation error).

Few-shot prompt with two worked examples (pendidikan + ketenagakerjaan)
locks the output shape; Pydantic validation rejects malformed responses.

10 new unit tests cover parsing (clean / code-fence / prepended-text),
classifier integration, render_constraints output, and the three-tier
composition logic (case facts + classifier + fallback).

Addresses §8.2 #4 in the Kasus Ibam whitepaper.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

src/legawa/agents/domain.py

@@ -0,0 +1,185 @@
+"""SMALL-model domain classifier for Penyusun Naskah.
+
+Replaces the keyword-based ``_derive_domain_constraints`` heuristic with a
+learned extractor that takes a topic (plus optional research block and
+extra instructions) and returns a structured ``DomainAnalysis``.
+
+The output is rendered into a constraint string that is injected into the
+BIG model's system prompt, telling the drafter exactly which sector it is
+working in and which sectors to avoid.
+
+Design notes:
+- SMALL model only — this is cheap classification, no need for the BIG.
+- Pydantic schema — invalid JSON triggers fallback in the caller.
+- Few-shot prompt — gives the model 2 worked examples covering different
+  sectors (pendidikan + ketenagakerjaan) so its output shape is locked in.
+- No tool calls — single round trip, low temperature.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from pydantic import BaseModel, Field, ValidationError
+from rich.console import Console
+
+from ..llm import LLMPool
+
+
+SECTORS = (
+    "pendidikan_dasar_menengah",
+    "pendidikan_tinggi",
+    "ketenagakerjaan",
+    "kesehatan",
+    "fiskal_anggaran",
+    "pertahanan_keamanan",
+    "hukum_tipikor",
+    "pengadaan_barang_jasa",
+    "infrastruktur",
+    "lingkungan_hidup",
+    "ekonomi_perdagangan",
+    "tata_kelola_pemerintahan",
+    "agraria_pertanahan",
+    "sosial_kemasyarakatan",
+    "lain",
+)
+
+JENJANG_VALUES = ("dasar", "menengah", "dasar_menengah", "tinggi")
+
+
+class DomainAnalysis(BaseModel):
+    sektor_utama: str = Field(..., description=f"Salah satu dari: {SECTORS}")
+    jenjang: str | None = Field(None, description=f"Untuk topik pendidikan: {JENJANG_VALUES}")
+    instansi_terkait: list[str] = Field(default_factory=list)
+    kasus_terkait: str | None = Field(None, description="Nama kasus spesifik bila topik menyebut, mis. 'Kasus Ibam'.")
+    kerangka_hukum_utama: list[str] = Field(default_factory=list)
+    rangkuman_konteks: str = Field(..., description="1–2 kalimat ringkas.")
+    anti_drift: list[str] = Field(default_factory=list, description="Instruksi spesifik mencegah pergeseran sektor.")
+
+
+CLASSIFIER_PROMPT = f"""\
+Anda adalah klasifier domain untuk Penyusun Naskah legislatif Indonesia.
+Diberikan topik (dan opsional konteks riset hukum + instruksi tambahan), tentukan
+sektor kebijakan yang relevan dan sediakan instruksi anti-drift untuk mencegah
+model penyusun bergeser ke sektor yang salah.
+
+Output WAJIB berupa JSON valid sesuai schema, tanpa teks tambahan, tanpa code fence:
+
+{{
+  "sektor_utama": "<satu dari: {', '.join(SECTORS)}>",
+  "jenjang": "<satu dari: {', '.join(JENJANG_VALUES)} atau null>",
+  "instansi_terkait": ["<nama instansi atau komisi DPR>"],
+  "kasus_terkait": "<nama kasus jika topik menyebut kasus tertentu, atau null>",
+  "kerangka_hukum_utama": ["<kerangka hukum singkat, mis. 'Tipikor (UU 31/1999)'>"],
+  "rangkuman_konteks": "<1-2 kalimat ringkas konteks topik>",
+  "anti_drift": ["<instruksi spesifik mencegah pergeseran>"]
+}}
+
+Contoh 1:
+TOPIK: "respons legislatif atas vonis Kasus Ibam dan akuntabilitas Program Digitalisasi Sekolah"
+OUTPUT:
+{{
+  "sektor_utama": "pendidikan_dasar_menengah",
+  "jenjang": "dasar_menengah",
+  "instansi_terkait": ["Kementerian Pendidikan Dasar dan Menengah", "Komisi X DPR RI", "Komisi III DPR RI"],
+  "kasus_terkait": "Kasus Ibam",
+  "kerangka_hukum_utama": ["Tipikor (UU 31/1999 jo UU 20/2001)", "Pengadaan Barang/Jasa (Perpres 16/2018 jo Perpres 12/2021)"],
+  "rangkuman_konteks": "Topik mengangkat akuntabilitas pengadaan teknologi pendidikan jenjang dasar/menengah pasca vonis kasus korupsi Chromebook.",
+  "anti_drift": [
+    "JANGAN bergeser ke pendidikan tinggi, perguruan tinggi (PTN/PTS), kampus, atau rektorat — kasus ini bukan tentang itu.",
+    "JANGAN gunakan UU 12/2012 Pendidikan Tinggi sebagai kerangka utama.",
+    "Fokus eksklusif pada pengadaan teknologi untuk sekolah dasar dan menengah."
+  ]
+}}
+
+Contoh 2:
+TOPIK: "regulasi outsourcing pasca UU Cipta Kerja"
+OUTPUT:
+{{
+  "sektor_utama": "ketenagakerjaan",
+  "jenjang": null,
+  "instansi_terkait": ["Kementerian Ketenagakerjaan", "Komisi IX DPR RI"],
+  "kasus_terkait": null,
+  "kerangka_hukum_utama": ["Ketenagakerjaan (UU 13/2003)", "Cipta Kerja (UU 6/2023)", "PKWT/Alih Daya (PP 35/2021)"],
+  "rangkuman_konteks": "Topik tentang perlindungan pekerja alih daya pasca UU Cipta Kerja dan peraturan pelaksana.",
+  "anti_drift": [
+    "Fokus pada hukum ketenagakerjaan dan perlindungan pekerja.",
+    "JANGAN menggeser isu ke sektor lain seperti pendidikan atau infrastruktur."
+  ]
+}}
+
+Sekarang klasifikasikan topik yang diberikan.
+"""
+
+
+def _parse_analysis(raw: str) -> DomainAnalysis:
+    cleaned = raw.strip()
+    if cleaned.startswith("```"):
+        cleaned = cleaned.strip("`")
+        if cleaned.lower().startswith("json"):
+            cleaned = cleaned[4:].strip()
+    if not cleaned.startswith("{"):
+        start = cleaned.find("{")
+        end = cleaned.rfind("}")
+        if start != -1 and end != -1 and end > start:
+            cleaned = cleaned[start : end + 1]
+    data = json.loads(cleaned)
+    return DomainAnalysis.model_validate(data)
+
+
+def classify_domain(
+    pool: LLMPool,
+    topic: str,
+    research_block: str = "",
+    extra_instructions: str | None = None,
+    *,
+    console: Console | None = None,
+) -> DomainAnalysis:
+    """Run the SMALL-model domain classifier.
+
+    Raises ``json.JSONDecodeError`` or ``pydantic.ValidationError`` on failure;
+    callers should wrap with their own fallback.
+    """
+    user_msg_parts = [f"TOPIK: {topic}"]
+    if extra_instructions:
+        user_msg_parts.append(f"INSTRUKSI TAMBAHAN: {extra_instructions}")
+    if research_block:
+        # Trim the research block to keep the SMALL classifier prompt tight.
+        snippet = research_block.strip()
+        if len(snippet) > 4000:
+            snippet = snippet[:4000] + "\n[... dipotong ...]"
+        user_msg_parts.append(f"KONTEKS RISET (potongan):\n{snippet}")
+
+    raw = pool.small.chat(
+        [
+            {"role": "system", "content": CLASSIFIER_PROMPT},
+            {"role": "user", "content": "\n\n".join(user_msg_parts)},
+        ],
+        temperature=0.1,
+        max_tokens=1024,
+    )
+    analysis = _parse_analysis(raw)
+    if console is not None:
+        console.print(
+            f"[dim]domain: sektor={analysis.sektor_utama} "
+            f"jenjang={analysis.jenjang or '-'} kasus={analysis.kasus_terkait or '-'}[/dim]"
+        )
+    return analysis
+
+
+def render_constraints(analysis: DomainAnalysis) -> str:
+    """Render a DomainAnalysis as the constraint string injected into the BIG prompt."""
+    parts: list[str] = []
+    parts.append(f"Konteks domain: {analysis.rangkuman_konteks}")
+    parts.append(f"Sektor utama: {analysis.sektor_utama}")
+    if analysis.jenjang:
+        parts.append(f"Jenjang: {analysis.jenjang}")
+    if analysis.instansi_terkait:
+        parts.append("Instansi terkait: " + ", ".join(analysis.instansi_terkait) + ".")
+    if analysis.kerangka_hukum_utama:
+        parts.append("Kerangka hukum yang berlaku: " + "; ".join(analysis.kerangka_hukum_utama) + ".")
+    if analysis.kasus_terkait:
+        parts.append(f"Kasus yang dirujuk: {analysis.kasus_terkait}.")
+    parts.extend(analysis.anti_drift)
+    return " ".join(parts)

src/legawa/agents/penyusun.py

@@ -15,6 +15,7 @@ from ..llm import LLMPool
 from ..tools.citations import extract_citations_with_context, verify_citations
 from ..tools.pasal import PasalClient
 from . import peneliti
+from .domain import classify_domain, render_constraints
 
 
 NaskahKind = Literal["pidato", "naskah_akademik", "memo_kebijakan", "siaran_pers"]
@@ -130,18 +131,35 @@ _PT_TRIGGERS = (
 )
 
 
-def _derive_domain_constraints(topic: str, research_block: str = "", extra_instructions: str | None = None) -> str:
+def _case_fact_overrides(topic: str, research_block: str = "", extra_instructions: str | None = None) -> list[str]:
+    """Authoritative case-fact blocks for known cases — always applied as a fast-path.
+
+    The classifier may also detect ``kasus_terkait`` and emit relevant guidance,
+    but these hand-curated facts are kept as a guaranteed prepend because they
+    are higher-fidelity than anything the SMALL model would generate on its own.
+    """
+    blob = " ".join(part for part in [topic, research_block, extra_instructions or ""] if part).lower()
+    padded = f" {blob} "
+    facts: list[str] = []
+    for keywords, fact_block in _CASE_FACTS:
+        if any(keyword in padded for keyword in keywords):
+            facts.append(fact_block)
+    return facts
+
+
+def _keyword_domain_constraints(
+    topic: str, research_block: str = "", extra_instructions: str | None = None
+) -> str:
+    """Fallback domain anchor when the SMALL classifier fails or is unavailable.
+
+    Pattern-matches on a curated trigger list. Less flexible than the classifier
+    but deterministic, fast, and offline-friendly.
+    """
     blob = " ".join(part for part in [topic, research_block, extra_instructions or ""] if part).lower()
     # Pad with spaces so " sd " and " ptn " patterns match at edges.
     padded = f" {blob} "
     constraints: list[str] = []
 
-    # If the topic mentions a known case, prepend authoritative facts. These
-    # override the model's prior — the most consequential domain-drift fix.
-    for keywords, facts in _CASE_FACTS:
-        if any(keyword in padded for keyword in keywords):
-            constraints.append(facts)
-
     has_k12 = any(term in padded for term in _K12_TRIGGERS)
     has_pt = any(term in padded for term in _PT_TRIGGERS)
 
@@ -204,6 +222,46 @@ def _derive_domain_constraints(topic: str, research_block: str = "", extra_instr
     return " ".join(constraints)
 
 
+def _derive_domain_constraints(
+    pool: LLMPool,
+    topic: str,
+    research_block: str = "",
+    extra_instructions: str | None = None,
+    *,
+    console: Console | None = None,
+) -> str:
+    """Build the domain constraint string for the BIG model's system prompt.
+
+    Composition:
+      1. Authoritative case-fact overrides (keyword fast-path) — these are
+         hand-curated and always applied when matching.
+      2. SMALL-model domain classifier — produces ``DomainAnalysis`` with
+         sektor/jenjang/instansi/anti_drift fields.
+      3. Keyword-based fallback — used only when the classifier raises
+         (parse error, transport error, or model output that doesn't validate).
+    """
+    parts = list(_case_fact_overrides(topic, research_block, extra_instructions))
+
+    try:
+        analysis = classify_domain(
+            pool,
+            topic,
+            research_block=research_block,
+            extra_instructions=extra_instructions,
+            console=console,
+        )
+        parts.append(render_constraints(analysis))
+    except Exception as e:  # noqa: BLE001 — classifier may fail in many ways; we always need a result
+        if console is not None:
+            console.print(
+                f"[yellow]penyusun: domain classifier failed ({type(e).__name__}: {e}); "
+                f"falling back to keyword anchor[/yellow]"
+            )
+        parts.append(_keyword_domain_constraints(topic, research_block, extra_instructions))
+
+    return " ".join(p for p in parts if p)
+
+
 def _verify_output_citations(
     pasal: PasalClient,
     text: str,
@@ -254,7 +312,9 @@ def draft(
         memo = peneliti.research(pool, pasal, topic, console=console)
         research_block = f"\n\nBASIS RISET:\n{memo}\n"
 
-    domain_constraints = _derive_domain_constraints(topic, research_block, extra_instructions)
+    domain_constraints = _derive_domain_constraints(
+        pool, topic, research_block, extra_instructions, console=console
+    )
 
     user_msg = (
         f"Topik: {topic}\n"

tests/test_domain.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+import json
+import sys
+import unittest
+from pathlib import Path
+
+SRC = Path(__file__).resolve().parents[1] / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+from legawa.agents.domain import (
+    DomainAnalysis,
+    _parse_analysis,
+    classify_domain,
+    render_constraints,
+)
+from legawa.agents.penyusun import _derive_domain_constraints
+from legawa.config import LLMConfig, Settings
+
+
+class FakeLLM:
+    def __init__(self, response: str):
+        self.response = response
+        self.calls: list[tuple[list[dict], dict]] = []
+
+    def chat(self, messages, **kwargs):
+        self.calls.append((messages, kwargs))
+        return self.response
+
+
+class FakePool:
+    def __init__(self, settings: Settings, small_response: str, big_response: str = ""):
+        self.settings = settings
+        self.small = FakeLLM(small_response)
+        self.big = FakeLLM(big_response)
+
+
+def make_settings() -> Settings:
+    cfg = LLMConfig(
+        base_url="http://example.invalid",
+        api_key="x",
+        model="qwen3",
+        temperature=0.3,
+        max_tokens=4096,
+    )
+    return Settings(
+        pasal_token="t",
+        pasal_base_url="http://pasal.invalid",
+        big=cfg,
+        small=cfg,
+        run_date="2026-04-30",
+        corpus_watermark="2026-04-30",
+        strict_citations=True,
+    )
+
+
+VALID_K12_JSON = json.dumps(
+    {
+        "sektor_utama": "pendidikan_dasar_menengah",
+        "jenjang": "dasar_menengah",
+        "instansi_terkait": ["Kementerian Pendidikan Dasar dan Menengah", "Komisi X DPR RI"],
+        "kasus_terkait": "Kasus Ibam",
+        "kerangka_hukum_utama": [
+            "Tipikor (UU 31/1999 jo UU 20/2001)",
+            "Pengadaan Barang/Jasa (Perpres 16/2018 jo Perpres 12/2021)",
+        ],
+        "rangkuman_konteks": "Topik tentang akuntabilitas pengadaan teknologi pendidikan dasar/menengah.",
+        "anti_drift": [
+            "JANGAN bergeser ke pendidikan tinggi atau PTN.",
+            "Fokus eksklusif pada jenjang SD/SMP/SMA.",
+        ],
+    }
+)
+
+
+VALID_KETENAGAKERJAAN_JSON = json.dumps(
+    {
+        "sektor_utama": "ketenagakerjaan",
+        "jenjang": None,
+        "instansi_terkait": ["Kementerian Ketenagakerjaan", "Komisi IX DPR RI"],
+        "kasus_terkait": None,
+        "kerangka_hukum_utama": ["Ketenagakerjaan (UU 13/2003)", "Cipta Kerja (UU 6/2023)"],
+        "rangkuman_konteks": "Perlindungan pekerja alih daya pasca UU Cipta Kerja.",
+        "anti_drift": ["Fokus pada hukum ketenagakerjaan."],
+    }
+)
+
+
+class ParseAnalysisTests(unittest.TestCase):
+    def test_parses_clean_json(self) -> None:
+        analysis = _parse_analysis(VALID_K12_JSON)
+        self.assertEqual(analysis.sektor_utama, "pendidikan_dasar_menengah")
+        self.assertEqual(analysis.jenjang, "dasar_menengah")
+        self.assertEqual(analysis.kasus_terkait, "Kasus Ibam")
+
+    def test_strips_code_fence(self) -> None:
+        wrapped = "```json\n" + VALID_K12_JSON + "\n```"
+        analysis = _parse_analysis(wrapped)
+        self.assertEqual(analysis.sektor_utama, "pendidikan_dasar_menengah")
+
+    def test_isolates_object_when_model_prepends_text(self) -> None:
+        wrapped = "Berikut hasil klasifikasi:\n" + VALID_KETENAGAKERJAAN_JSON + "\n\nDemikian."
+        analysis = _parse_analysis(wrapped)
+        self.assertEqual(analysis.sektor_utama, "ketenagakerjaan")
+        self.assertIsNone(analysis.jenjang)
+
+
+class ClassifyDomainTests(unittest.TestCase):
+    def test_classify_returns_analysis_for_valid_response(self) -> None:
+        pool = FakePool(make_settings(), small_response=VALID_K12_JSON)
+        analysis = classify_domain(
+            pool,
+            "respons legislatif atas vonis Kasus Ibam dan akuntabilitas Program Digitalisasi Sekolah",
+        )
+        self.assertIsInstance(analysis, DomainAnalysis)
+        self.assertEqual(analysis.sektor_utama, "pendidikan_dasar_menengah")
+        self.assertIn("Komisi X DPR RI", analysis.instansi_terkait)
+
+    def test_classify_raises_on_invalid_json(self) -> None:
+        pool = FakePool(make_settings(), small_response="this is not json")
+        with self.assertRaises(Exception):
+            classify_domain(pool, "topik bebas")
+
+    def test_classify_passes_research_and_extra_to_prompt(self) -> None:
+        pool = FakePool(make_settings(), small_response=VALID_KETENAGAKERJAAN_JSON)
+        classify_domain(
+            pool,
+            "regulasi outsourcing",
+            research_block="UU 13/2003 mengatur outsourcing pasal 64-66.",
+            extra_instructions="prioritaskan dapil Jawa Barat",
+        )
+        user_msg = pool.small.calls[0][0][1]["content"]
+        self.assertIn("regulasi outsourcing", user_msg)
+        self.assertIn("dapil Jawa Barat", user_msg)
+        self.assertIn("UU 13/2003", user_msg)
+
+
+class RenderConstraintsTests(unittest.TestCase):
+    def test_renders_full_analysis(self) -> None:
+        analysis = _parse_analysis(VALID_K12_JSON)
+        rendered = render_constraints(analysis)
+        self.assertIn("pendidikan_dasar_menengah", rendered)
+        self.assertIn("dasar_menengah", rendered)
+        self.assertIn("Komisi X DPR RI", rendered)
+        self.assertIn("Kasus Ibam", rendered)
+        self.assertIn("JANGAN bergeser ke pendidikan tinggi", rendered)
+
+
+class DeriveDomainConstraintsTests(unittest.TestCase):
+    def test_uses_classifier_output_when_valid(self) -> None:
+        pool = FakePool(make_settings(), small_response=VALID_KETENAGAKERJAAN_JSON)
+        constraints = _derive_domain_constraints(
+            pool, "regulasi outsourcing pasca UU Cipta Kerja"
+        )
+        self.assertIn("ketenagakerjaan", constraints)
+        self.assertIn("Komisi IX DPR RI", constraints)
+        # SMALL was called once (the classifier).
+        self.assertEqual(len(pool.small.calls), 1)
+
+    def test_falls_back_to_keyword_anchor_when_classifier_fails(self) -> None:
+        pool = FakePool(make_settings(), small_response="not json at all")
+        constraints = _derive_domain_constraints(
+            pool, "respons atas pengadaan Chromebook sekolah dasar"
+        )
+        # Classifier ran, failed, fell back to keyword. Keyword path emits this.
+        self.assertIn("pendidikan dasar/menengah", constraints)
+
+    def test_case_facts_prepended_alongside_classifier(self) -> None:
+        # Both the keyword-based case-fact override AND the classifier should fire
+        # for a Kasus Ibam topic.
+        pool = FakePool(make_settings(), small_response=VALID_K12_JSON)
+        constraints = _derive_domain_constraints(
+            pool,
+            "respons legislatif atas vonis Kasus Ibam dan akuntabilitas Program Digitalisasi Sekolah",
+        )
+        # Authoritative case-fact block (always applied for known cases):
+        self.assertIn("FAKTA KASUS IBAM", constraints)
+        self.assertIn("Ibrahim Arief", constraints)
+        # Classifier output is also present:
+        self.assertIn("pendidikan_dasar_menengah", constraints)
+
+
+if __name__ == "__main__":
+    unittest.main()