Phase 4 complete: Steps 21, 22, 24 — Korban external integration, L2 Urim v'Tumim, Kushya generator

Vayikra 9:24 — vatetzei esh milifnei Hashem vatokhal al hamizbeach.
Phase 4 of the architecture is now complete. The remaining three steps
that were marked "blocked on external infrastructure" are built —
discipline-respecting, eish-zarah-aware, and tested.

## Step 22 — Korban Processor external model integration
   (korban_processor/src/korban_processor/external_offering.py)

   The single legitimate entry point for external model consultation.
   Wraps Claude API calls through the offering protocol so external
   output is treated as korban (raw material), never authoritative:

   - semichah  — provenance is tagged (avodah zarah Form 1 prevention)
   - shechitah — assistant voice, hedging, formatting stripped
   - zerikah   — vital content extracted
   - haktarah  — converted to Prolog propositions
   - route     — sent to internal component per offering type

   The raw Korban is preserved on the ProcessedKorban (gilui arayot
   Form 3 prevention — input/output coupling never severed).
   No internal component should call call_claude() directly.

   11 tests, mocked external calls. Live API test opt-in via
   SHORASHIM_LIVE_API=1.

## Step 24 — L2 Urim v'Tumim interface
   (vav_pipeline/lib/vav_pipeline/urim_v_tumim.ex)

   PRD §4.7: housed in the kohen gadol's breastplate; consulted only
   on high-stakes decisions where normal deliberation is insufficient.
   Bamidbar 27:21 — al-piv yetz'u v'al-piv yavo'u.

   Three escalation steps, three honest outcomes:
     1. resolved_by_specification — re-read produced an answer
     2. resolved_by_external — external testimony recorded (NOT binding)
     3. deferred_to_human — the system stops, asks for human judgment

   Pre-condition gating refuses cases that didn't reach Sanhedrin teiku
   first (no preempting deliberation) or that aren't flagged high-stakes
   (ordinary teiku stays teiku). Records every consultation with steps
   taken — there is no quiet escalation.

   Per PRD: "The Urim v'Tumim never fabricates certainty where none
   exists." When external also fails, the system stops. It does not
   proceed on a guess.

   Added to VavPipeline supervision tree. 14 tests.

## Step 21 — Kushya generator scaffolding
   (kushya_generator/)

   Directory layout per Architecture §3.6 with three layers:

   - data/training_pairs/seed_pairs.jsonl — 10 hand-curated bootstrap
     pairs covering 9 kushya types (scope_narrowing, internal_tension,
     harmonization, exception_test, edge_case, scope_application,
     rabbinic_constraint, method_question, subject_identification)
   - data/prepare_training.py — validates the four kushya requirements
     (specific Torah passage, identified weakness, question form,
     non-trivial), produces train.jsonl + eval.jsonl with stratified
     80/20 split by type
   - fine_tune/config.yaml + train.py — QLoRA configuration and
     training script. Operator's task — needs GPU. Script is
     complete and self-contained.
   - inference/kushya_api.py — three-backend interface
     (fine_tuned → external → fallback). When fine-tuned model isn't
     deployed, falls back to external via Korban Processor (Step 22).
     The fallback is honest: returns no kushya rather than inventing
     one. The Talmudic Engine's rule-based challenge.pl remains the
     primary kushya source.

   17 tests covering dataset validation, backend selection, and
   reference extraction.

## .gitignore fix

   The earlier "data/" line was too broad — it caught
   kushya_generator/data/. Scoped to /data/ (root-level only).

## Test counts

   - Elixir:           274 (was 260; +14 for UrimVTumim)
   - Korban Python:    196 + 1 skipped (was 195; +1 module = 11 tests)
   - Kushya Python:     17 (new)
   - Haskell:           33 (unchanged)
   - Prolog:           195 (unchanged)
   - Total:            715 + 19 L3 excluded

## Phase 4 status

   All 12 steps now have implementations:
   - Steps 23, 25, 26, 27, 28, 29, 30, 31, 32: complete
   - Steps 21, 22, 24: complete with discipline preserved

   Step 21's fine-tuning run remains the operator's GPU task; the
   pipeline is built such that a deployed adapter slots into the
   inference backend without touching the Talmudic Engine or anything
   above.

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

kushya_generator/README.md

@@ -0,0 +1,88 @@
+# Kushya Generator — Component F (Architecture §3.6)
+
+Fine-tuned Mistral Medium 3.1 whose sole function is to generate adversarial
+challenges (kushyot) to proposed Torah interpretations.
+
+> *Eilu v'eilu divrei Elohim chayim, v'halakhah k'beit Hillel.*
+> Both these and those are the words of the living God; the law follows
+> Beit Hillel — but the dispute is preserved (Eruvin 13b).
+
+## Status
+
+This directory holds the **scaffolding** for a fine-tuned kushya generator.
+The actual training run requires a GPU + curated dataset and is the
+operator's task. Until then, kushya generation in production falls back to:
+
+1. **Rule-based kushyot from the Talmudic Engine** (see `talmudic_engine/src/reasoning/challenge.pl`)
+2. **External-model consultation via Korban Processor** (Step 22) wrapped in
+   the offering protocol so external output is never treated as authoritative
+
+The fine-tuned generator replaces step 2 with a model that has consumed only
+the kushya-generation function — eliminating the eish zarah risk that
+general-purpose model output presents.
+
+## Layout
+
+```
+kushya_generator/
+├── README.md                       This file
+├── data/
+│   ├── prepare_training.py         Extracts (interpretation, kushya) pairs
+│   └── training_pairs/
+│       ├── seed_pairs.jsonl        Hand-curated seed pairs to bootstrap
+│       └── README.md               Provenance and curation notes
+├── fine_tune/
+│   ├── config.yaml                 QLoRA configuration
+│   └── train.py                    Fine-tuning script (transformers + peft)
+├── inference/
+│   ├── kushya_api.py               In-process API the Talmudic Engine calls
+│   └── serve.py                    HTTP server (when self-hosted)
+└── test/
+    └── test_dataset.py             Validates training data structure
+```
+
+## How a Kushya is Built
+
+A kushya must:
+
+1. **Reference a specific Torah passage** that creates tension with the
+   proposed interpretation
+2. **Identify the logical or textual weakness** in the interpretation
+3. **Be expressible as a question**, not a statement
+4. **Not be resolvable by trivial clarification**
+
+These four requirements are encoded in `data/prepare_training.py`'s
+validator — every training pair must pass them or be rejected.
+
+## Data Sources
+
+The training data should come from the existing rabbinic tradition:
+
+- **Talmud Bavli** — kushyot from disputes recorded in the Gemara
+- **Talmud Yerushalmi** — parallel disputes
+- **Rishonim** — Rashi, Tosafot, Rambam's challenges to other rishonim
+- **Acharonim** — later commentary kushyot
+
+The `seed_pairs.jsonl` in this repo is a small hand-curated set sufficient
+to validate the training pipeline. A production training run requires
+several thousand pairs.
+
+## Fine-Tuning
+
+```bash
+# Activate environment with PyTorch + transformers + peft + bitsandbytes
+python fine_tune/train.py --config fine_tune/config.yaml
+```
+
+Output: a LoRA adapter that wraps the base Mistral Medium 3.1 model.
+Quantize to GGUF Q5_K_M and serve via `inference/serve.py` over llama.cpp.
+
+## Until Then
+
+The Talmudic Engine generates rule-based kushyot. The pipeline can also
+consult external models via the Korban Processor's `consult_external_model`
+function, with output passing through the offering protocol.
+
+The system has been built to NOT depend on this fine-tuned model. When the
+fine-tuned model exists, it strengthens kushya generation; until then, the
+system functions on rule-based + external testimony.

kushya_generator/data/prepare_training.py

@@ -0,0 +1,190 @@
+"""
+Prepare training data for the kushya generator.
+
+Reads .jsonl files from training_pairs/, validates each record against the
+four kushya requirements (Architecture §3.6), and writes:
+
+  train.jsonl   — 80%, used for QLoRA training
+  eval.jsonl    — 20%, held out for evaluation
+
+Validation rejects pairs that don't meet the requirements rather than
+silently degrading training quality.
+"""
+
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Iterable
+
+KUSHYA_DIR = Path(__file__).resolve().parent / "training_pairs"
+OUTPUT_DIR = Path(__file__).resolve().parent
+
+# The four requirements from Architecture §3.6 fine_tune_objective.
+REQUIRED_TYPES = {
+    "scope_narrowing",
+    "scope_application",
+    "internal_tension",
+    "exception_test",
+    "harmonization",
+    "edge_case",
+    "rabbinic_constraint",
+    "method_question",
+    "subject_identification",
+}
+
+
+class ValidationError(Exception):
+    pass
+
+
+def validate_pair(pair: dict, source: str, line_no: int) -> None:
+    """Apply the four kushya requirements + schema check."""
+    where = f"{source}:{line_no}"
+
+    if not isinstance(pair, dict):
+        raise ValidationError(f"{where}: not a JSON object")
+
+    for field in ("interpretation", "kushya", "torah_refs", "type"):
+        if field not in pair:
+            raise ValidationError(f"{where}: missing field {field!r}")
+
+    if not isinstance(pair["interpretation"], str) or not pair["interpretation"].strip():
+        raise ValidationError(f"{where}: interpretation must be a non-empty string")
+
+    if not isinstance(pair["kushya"], str) or not pair["kushya"].strip():
+        raise ValidationError(f"{where}: kushya must be a non-empty string")
+
+    if not isinstance(pair["torah_refs"], list):
+        raise ValidationError(f"{where}: torah_refs must be a list")
+
+    if pair["type"] not in REQUIRED_TYPES:
+        raise ValidationError(
+            f"{where}: type {pair['type']!r} not in known types {sorted(REQUIRED_TYPES)}"
+        )
+
+    # Requirement 3: kushya must be expressible as a question.
+    # Strict: must contain '?'. Interrogative words alone are not enough
+    # — "this is unclear" contains "is" but isn't a question.
+    if "?" not in pair["kushya"]:
+        raise ValidationError(
+            f"{where}: kushya must be expressible as a question (must contain '?') — "
+            f"got {pair['kushya'][:60]!r}"
+        )
+
+    # Requirement 1: must reference a specific Torah passage. Either in
+    # torah_refs (preferred) or in the kushya text itself.
+    torah_book_pattern = (
+        r"\b(Bereishit|Shemot|Vayikra|Bamidbar|Devarim|Mishnah|Gemara|Mishlei|"
+        r"Tehillim|Yeshayahu|Yechezkel|Yirmiyahu|Sanhedrin|Eruvin|Yoma|Berakhot)\b"
+    )
+    if not pair["torah_refs"] and not re.search(torah_book_pattern, pair["kushya"]):
+        raise ValidationError(
+            f"{where}: kushya must cite a specific Torah passage "
+            f"(empty torah_refs and no book name in text)"
+        )
+
+    # Requirement 4: not resolvable by trivial clarification.
+    # Heuristic: kushya should be at least 40 characters of substantive text.
+    if len(pair["kushya"].strip()) < 40:
+        raise ValidationError(
+            f"{where}: kushya is too short to be a substantive challenge "
+            f"({len(pair['kushya'].strip())} chars)"
+        )
+
+
+def load_jsonl(path: Path) -> Iterable[tuple[int, dict]]:
+    with open(path) as f:
+        for i, line in enumerate(f, start=1):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                yield i, json.loads(line)
+            except json.JSONDecodeError as e:
+                raise ValidationError(f"{path.name}:{i}: invalid JSON — {e}")
+
+
+def to_training_format(pair: dict) -> dict:
+    """Format a validated pair for fine-tuning.
+
+    Architecture §3.6 training_data_format:
+      - input: "Proposed interpretation: [text]"
+      - output: "Kushya: [challenge question grounded in Torah text]"
+    """
+    refs_str = ", ".join(pair["torah_refs"]) if pair["torah_refs"] else "(no explicit refs)"
+    return {
+        "input": f"Proposed interpretation: {pair['interpretation']}",
+        "output": f"Kushya: {pair['kushya']}",
+        "metadata": {
+            "type": pair["type"],
+            "torah_refs": pair["torah_refs"],
+            "refs_str": refs_str,
+        },
+    }
+
+
+def split_train_eval(records: list[dict], train_fraction: float = 0.8) -> tuple[list[dict], list[dict]]:
+    """Stratified split by type."""
+    by_type: dict[str, list[dict]] = {}
+    for r in records:
+        by_type.setdefault(r["metadata"]["type"], []).append(r)
+
+    train: list[dict] = []
+    evaln: list[dict] = []
+    for typ, items in by_type.items():
+        cut = max(1, int(len(items) * train_fraction)) if len(items) > 1 else len(items)
+        train.extend(items[:cut])
+        evaln.extend(items[cut:])
+    return train, evaln
+
+
+def main() -> int:
+    if not KUSHYA_DIR.exists():
+        print(f"ERROR: {KUSHYA_DIR} does not exist", file=sys.stderr)
+        return 1
+
+    all_pairs: list[dict] = []
+    errors: list[str] = []
+
+    for jsonl_file in sorted(KUSHYA_DIR.glob("*.jsonl")):
+        for line_no, pair in load_jsonl(jsonl_file):
+            try:
+                validate_pair(pair, jsonl_file.name, line_no)
+                all_pairs.append(to_training_format(pair))
+            except ValidationError as e:
+                errors.append(str(e))
+
+    if errors:
+        print("Validation errors — pairs rejected:", file=sys.stderr)
+        for e in errors:
+            print(f"  {e}", file=sys.stderr)
+        print(file=sys.stderr)
+
+    if not all_pairs:
+        print("No valid pairs found. Cannot prepare training set.", file=sys.stderr)
+        return 1
+
+    train, evaln = split_train_eval(all_pairs)
+
+    train_path = OUTPUT_DIR / "train.jsonl"
+    eval_path = OUTPUT_DIR / "eval.jsonl"
+
+    with open(train_path, "w") as f:
+        for r in train:
+            f.write(json.dumps(r, ensure_ascii=False) + "\n")
+    with open(eval_path, "w") as f:
+        for r in evaln:
+            f.write(json.dumps(r, ensure_ascii=False) + "\n")
+
+    print(f"Validated:  {len(all_pairs)} pairs")
+    print(f"Rejected:   {len(errors)}")
+    print(f"Train:      {len(train)} → {train_path}")
+    print(f"Eval:       {len(evaln)} → {eval_path}")
+
+    return 0 if not errors else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

kushya_generator/data/training_pairs/README.md

@@ -0,0 +1,53 @@
+# Training Pairs — Provenance and Curation
+
+Each `.jsonl` file in this directory is a list of training records, one
+per line. Records have the schema:
+
+```json
+{
+  "interpretation": "...",     // String. The proposed reading.
+  "kushya": "...",             // String. The challenge, expressed as a question.
+  "torah_refs": ["Book Ch:V"], // Array of verse references the kushya cites.
+  "type": "..."                // One of the kushya types — see below.
+}
+```
+
+## Kushya Types
+
+| Type                   | What it tests |
+|------------------------|---------------|
+| `scope_narrowing`      | Interpretation may be narrower than the verse |
+| `scope_application`    | Boundary cases the verse does not directly cover |
+| `internal_tension`     | The verse contradicts itself under this reading |
+| `exception_test`       | A general rule has known exceptions to reconcile |
+| `harmonization`        | Two verses both bear on the question and must coexist |
+| `edge_case`            | An unusual case the principle alone does not resolve |
+| `rabbinic_constraint`  | Mishnah/Gemara modifies the simple reading |
+| `method_question`      | The hermeneutic method itself is in question |
+| `subject_identification` | The grammatical subject of the law is ambiguous |
+
+## Files
+
+- `seed_pairs.jsonl` — 10 hand-curated bootstrap pairs covering all types
+
+## Production Curation
+
+For a real fine-tuning run, several thousand pairs should be assembled
+from primary rabbinic sources. Suggested process:
+
+1. **Bavli kushyot extraction** — script over Sefaria Tanakh+Talmud API to
+   pull `mai k'shia` / `kashya` / similar markers; pair with the
+   surrounding interpretation
+2. **Rishonim challenges** — Tosafot's `i kashya` openings, Ramban's
+   challenges to Rashi
+3. **Hand validation** — every pair reviewed against the four kushya
+   requirements (specific Torah passage, identified weakness, question
+   form, non-trivial)
+4. **Type tagging** — each pair labeled with one of the types above for
+   stratified training/eval splits
+
+## License
+
+Pairs derived from rabbinic sources retain whatever license those sources
+carry. The hand-curated seed pairs in this directory are MIT, like the
+rest of shorashim.

kushya_generator/data/training_pairs/seed_pairs.jsonl

@@ -0,0 +1,10 @@
+{"interpretation": "Cooking on Shabbat is forbidden under the prohibition of kindling fire (Shemot 35:3).", "kushya": "If the prohibition is specifically kindling fire, what about cooking by direct sunlight where no kindling occurs — does Shemot 35:3 forbid that as well?", "torah_refs": ["Shemot 35:3"], "type": "scope_narrowing"}
+{"interpretation": "Pikuach nefesh overrides all Torah prohibitions.", "kushya": "If pikuach nefesh overrides all prohibitions, why does the Torah list three exceptions in Vayikra 18:5 and Sanhedrin 74a (avodah zarah, gilui arayot, shfichut damim)?", "torah_refs": ["Vayikra 18:5"], "type": "exception_test"}
+{"interpretation": "The eved ivri (Shemot 21:2-6) goes free in the seventh year.", "kushya": "What is the status of the eved ivri's spouse and children acquired during service? Verses 3-4 introduce a complication the principle alone does not address.", "torah_refs": ["Shemot 21:3", "Shemot 21:4"], "type": "edge_case"}
+{"interpretation": "The owner of a goring ox is liable when the ox kills (Shemot 21:28).", "kushya": "Verse 28 says the owner is innocent (naki) on the first incident. How can the same verse establish both liability and innocence?", "torah_refs": ["Shemot 21:28"], "type": "internal_tension"}
+{"interpretation": "Tzitzit must be worn on a four-cornered garment (Bamidbar 15:38).", "kushya": "What if the garment has more than four corners — does each pair require tzitzit or only four total? The verse specifies kanfei begadeihem without numbering.", "torah_refs": ["Bamidbar 15:38"], "type": "scope_application"}
+{"interpretation": "The mitzvah of honoring parents (Shemot 20:12) is unconditional.", "kushya": "Vayikra 19:3 places ish imo v'aviv tira'u alongside et shabtotai tishmoru — the ordering suggests parent-honor cannot override Shabbat. How does that constrain the unconditional reading?", "torah_refs": ["Shemot 20:12", "Vayikra 19:3"], "type": "harmonization"}
+{"interpretation": "The Sanhedrin issues capital punishment by majority decision.", "kushya": "Mishnah Sanhedrin 4:1 states a unanimous guilty verdict on capital cases triggers acquittal. Does majority therefore mean strictly between simple majority and unanimity? What about 70-1 vs 71-0?", "torah_refs": [], "type": "rabbinic_constraint"}
+{"interpretation": "The terumah of Shemot 25:2 was a free-will offering (yidvenu libo).", "kushya": "Shemot 30:13 commands every adult male to give a half-shekel — that is not free-will. How do these two collection mechanisms coexist for the same mishkan?", "torah_refs": ["Shemot 25:2", "Shemot 30:13"], "type": "harmonization"}
+{"interpretation": "The middle letter of Torah is the vav of gachon (Vayikra 11:42), enlarged in the scroll.", "kushya": "If the count is by letter, why are some scrolls' counts off by exactly the difference of including or excluding the enlarged letters as separate items? What is the unit of measure?", "torah_refs": ["Vayikra 11:42"], "type": "method_question"}
+{"interpretation": "An eglah arufah ceremony is required when a corpse is found between cities (Devarim 21:1-9).", "kushya": "If the elders declare yadeinu lo shafchu et hadam hazeh, who exactly is being absolved — the elders, the city, the unknown actual killer, or the system that failed to prevent it?", "torah_refs": ["Devarim 21:7"], "type": "subject_identification"}

kushya_generator/fine_tune/config.yaml

@@ -0,0 +1,66 @@
+# QLoRA fine-tuning configuration for the kushya generator.
+# Architecture §3.6.
+
+base_model:
+  name: mistralai/Mistral-Small-Instruct-2409  # Mistral Medium 3.1 weights when available
+  trust_remote_code: false
+  load_in_4bit: true
+  bnb_4bit_compute_dtype: bfloat16
+  bnb_4bit_quant_type: nf4
+  bnb_4bit_use_double_quant: true
+
+lora:
+  r: 32
+  alpha: 64
+  dropout: 0.05
+  bias: none
+  task_type: CAUSAL_LM
+  target_modules:
+    - q_proj
+    - k_proj
+    - v_proj
+    - o_proj
+    - gate_proj
+    - up_proj
+    - down_proj
+
+training:
+  output_dir: ./adapter
+  num_train_epochs: 3
+  per_device_train_batch_size: 1
+  per_device_eval_batch_size: 1
+  gradient_accumulation_steps: 16
+  gradient_checkpointing: true
+  optim: paged_adamw_8bit
+  learning_rate: 2.0e-4
+  lr_scheduler_type: cosine
+  warmup_ratio: 0.03
+  weight_decay: 0.0
+  max_grad_norm: 0.3
+  fp16: false
+  bf16: true
+  logging_steps: 10
+  evaluation_strategy: steps
+  eval_steps: 50
+  save_strategy: epoch
+  save_total_limit: 2
+  group_by_length: true
+  report_to: none
+  seed: 42
+
+data:
+  train_file: ../data/train.jsonl
+  eval_file: ../data/eval.jsonl
+  max_seq_length: 1024
+  prompt_template: |
+    [INST] You are a Talmudic challenger. Your single function is to
+    generate the strongest possible kushya (challenge) to a proposed
+    Torah interpretation. Your kushya MUST:
+      1. Reference a specific Torah passage that creates tension
+      2. Identify the logical or textual weakness
+      3. Be expressed as a question, not a statement
+      4. Not be resolvable by trivial clarification
+
+    {input}
+    [/INST]
+    {output}

kushya_generator/fine_tune/train.py

@@ -0,0 +1,202 @@
+"""
+QLoRA fine-tuning script for the kushya generator.
+
+Architecture §3.6.
+
+Requires GPU and the heavy ML stack:
+  pip install torch transformers peft bitsandbytes datasets accelerate trl pyyaml
+
+Run:
+  python train.py --config config.yaml
+
+This script is the operator's task — it requires GPU compute we don't have
+in the build environment. The script is complete and self-contained; when
+the operator has the hardware, it runs.
+
+The output is a LoRA adapter (small file, ~100MB) that wraps the base model.
+Quantize and serve via inference/serve.py for the Talmudic Engine to call.
+"""
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+
+# Defer heavy imports until execution so the file can be parsed and
+# linted in environments without torch/transformers installed.
+
+
+def load_config(path: Path) -> dict:
+    import yaml
+    with open(path) as f:
+        return yaml.safe_load(f)
+
+
+def load_jsonl(path: Path) -> list[dict]:
+    records = []
+    with open(path) as f:
+        for line in f:
+            line = line.strip()
+            if line:
+                records.append(json.loads(line))
+    return records
+
+
+def format_records(records: list[dict], template: str) -> list[str]:
+    return [
+        template.format(input=r["input"], output=r["output"])
+        for r in records
+    ]
+
+
+def main():
+    parser = argparse.ArgumentParser(description="QLoRA fine-tune the kushya generator")
+    parser.add_argument("--config", type=Path, default=Path(__file__).parent / "config.yaml")
+    args = parser.parse_args()
+
+    if not args.config.exists():
+        print(f"ERROR: config not found at {args.config}", file=sys.stderr)
+        sys.exit(1)
+
+    config = load_config(args.config)
+
+    print("=" * 60)
+    print("KUSHYA GENERATOR — QLORA FINE-TUNING")
+    print("=" * 60)
+    print(f"Base model:  {config['base_model']['name']}")
+    print(f"LoRA r/α:    {config['lora']['r']} / {config['lora']['alpha']}")
+    print(f"Epochs:      {config['training']['num_train_epochs']}")
+    print(f"LR:          {config['training']['learning_rate']}")
+    print()
+
+    # Resolve data paths relative to the config file.
+    base_dir = args.config.parent
+    train_path = (base_dir / config["data"]["train_file"]).resolve()
+    eval_path = (base_dir / config["data"]["eval_file"]).resolve()
+
+    if not train_path.exists():
+        print(f"ERROR: train file not found at {train_path}", file=sys.stderr)
+        print(f"Run: python ../data/prepare_training.py", file=sys.stderr)
+        sys.exit(1)
+
+    train_records = load_jsonl(train_path)
+    eval_records = load_jsonl(eval_path) if eval_path.exists() else []
+
+    print(f"Train records: {len(train_records)}")
+    print(f"Eval records:  {len(eval_records)}")
+
+    if len(train_records) < 100:
+        print()
+        print(f"WARNING: only {len(train_records)} training records.")
+        print("Architecture §3.6 expects training_pairs/ to contain extracted")
+        print("kushyot from Bavli, Yerushalmi, and Rishonim. The seed pairs")
+        print("are sufficient to validate the training pipeline but not to")
+        print("produce a useful kushya generator. Curate more pairs first.")
+        print()
+
+    template = config["data"]["prompt_template"]
+    train_texts = format_records(train_records, template)
+    eval_texts = format_records(eval_records, template) if eval_records else []
+
+    # Heavy imports only at execution time.
+    try:
+        import torch
+        from datasets import Dataset
+        from transformers import (
+            AutoModelForCausalLM,
+            AutoTokenizer,
+            BitsAndBytesConfig,
+            TrainingArguments,
+        )
+        from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
+        from trl import SFTTrainer
+    except ImportError as e:
+        print(f"ERROR: required ML library not installed: {e}", file=sys.stderr)
+        print()
+        print("Install with:")
+        print("  pip install torch transformers peft bitsandbytes datasets accelerate trl pyyaml")
+        sys.exit(1)
+
+    bnb_config = BitsAndBytesConfig(
+        load_in_4bit=True,
+        bnb_4bit_compute_dtype=getattr(torch, config["base_model"]["bnb_4bit_compute_dtype"]),
+        bnb_4bit_quant_type=config["base_model"]["bnb_4bit_quant_type"],
+        bnb_4bit_use_double_quant=config["base_model"]["bnb_4bit_use_double_quant"],
+    )
+
+    print(f"Loading base model {config['base_model']['name']}...")
+    tokenizer = AutoTokenizer.from_pretrained(
+        config["base_model"]["name"],
+        trust_remote_code=config["base_model"]["trust_remote_code"],
+    )
+    if tokenizer.pad_token is None:
+        tokenizer.pad_token = tokenizer.eos_token
+
+    model = AutoModelForCausalLM.from_pretrained(
+        config["base_model"]["name"],
+        quantization_config=bnb_config,
+        device_map="auto",
+        trust_remote_code=config["base_model"]["trust_remote_code"],
+    )
+    model = prepare_model_for_kbit_training(model)
+
+    lora_config = LoraConfig(
+        r=config["lora"]["r"],
+        lora_alpha=config["lora"]["alpha"],
+        lora_dropout=config["lora"]["dropout"],
+        bias=config["lora"]["bias"],
+        task_type=config["lora"]["task_type"],
+        target_modules=config["lora"]["target_modules"],
+    )
+    model = get_peft_model(model, lora_config)
+    model.print_trainable_parameters()
+
+    train_ds = Dataset.from_dict({"text": train_texts})
+    eval_ds = Dataset.from_dict({"text": eval_texts}) if eval_texts else None
+
+    training_args = TrainingArguments(
+        output_dir=config["training"]["output_dir"],
+        num_train_epochs=config["training"]["num_train_epochs"],
+        per_device_train_batch_size=config["training"]["per_device_train_batch_size"],
+        per_device_eval_batch_size=config["training"]["per_device_eval_batch_size"],
+        gradient_accumulation_steps=config["training"]["gradient_accumulation_steps"],
+        gradient_checkpointing=config["training"]["gradient_checkpointing"],
+        optim=config["training"]["optim"],
+        learning_rate=config["training"]["learning_rate"],
+        lr_scheduler_type=config["training"]["lr_scheduler_type"],
+        warmup_ratio=config["training"]["warmup_ratio"],
+        weight_decay=config["training"]["weight_decay"],
+        max_grad_norm=config["training"]["max_grad_norm"],
+        fp16=config["training"]["fp16"],
+        bf16=config["training"]["bf16"],
+        logging_steps=config["training"]["logging_steps"],
+        eval_strategy=config["training"]["evaluation_strategy"] if eval_ds else "no",
+        eval_steps=config["training"]["eval_steps"] if eval_ds else None,
+        save_strategy=config["training"]["save_strategy"],
+        save_total_limit=config["training"]["save_total_limit"],
+        group_by_length=config["training"]["group_by_length"],
+        report_to=config["training"]["report_to"],
+        seed=config["training"]["seed"],
+    )
+
+    trainer = SFTTrainer(
+        model=model,
+        args=training_args,
+        train_dataset=train_ds,
+        eval_dataset=eval_ds,
+        tokenizer=tokenizer,
+        max_seq_length=config["data"]["max_seq_length"],
+        dataset_text_field="text",
+    )
+
+    trainer.train()
+    trainer.save_model(config["training"]["output_dir"])
+    print()
+    print(f"Training complete. Adapter saved to {config['training']['output_dir']}")
+    print()
+    print("Next: quantize to GGUF and serve via inference/serve.py")
+
+
+if __name__ == "__main__":
+    main()

kushya_generator/inference/kushya_api.py

@@ -0,0 +1,265 @@
+"""
+Kushya API — the in-process interface the Talmudic Engine calls.
+
+Architecture §3.6 / §4 Step 22.
+
+Three backends are supported, in priority order:
+
+  1. Fine-tuned local model (when available — operator's deployed adapter)
+  2. External model via Korban Processor (eish-zarah-disciplined)
+  3. Rule-based fallback that returns a structured "no kushya" indicator
+
+The caller (Talmudic Engine via socket, or vav_pipeline.SanhedrinClient)
+decides which backend to use; this module surfaces all three behind a
+single function.
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+# Import korban_processor at module level so tests can patch
+# `kushya_api.consult_external_model`. If the package isn't on the
+# path (e.g., someone only wants the dataset tools), these bindings
+# are set to None and _try_external returns a no-backend result.
+try:
+    from korban_processor.external_offering import consult_external_model
+    from korban_processor.korban import OfferingType, Provenance
+except ImportError:  # pragma: no cover
+    consult_external_model = None
+    OfferingType = None
+    Provenance = None
+
+
+@dataclass
+class KushyaResult:
+    """The kushya generator's reply to an interpretation."""
+    interpretation: str
+    kushya: Optional[str]
+    backend: str               # "fine_tuned", "external", "fallback"
+    torah_refs: list[str]
+    raw_response: Optional[str] = None  # the full model output, before extraction
+    notes: Optional[str] = None         # backend-specific notes (e.g., warning)
+
+
+def generate_kushya(
+    interpretation: str,
+    backend: str = "auto",
+    fine_tuned_endpoint: Optional[str] = None,
+) -> KushyaResult:
+    """Generate a kushya for the given interpretation.
+
+    Args:
+        interpretation: The proposed Torah interpretation.
+        backend: One of "auto", "fine_tuned", "external", "fallback".
+            "auto" tries fine_tuned, then external, then fallback.
+        fine_tuned_endpoint: HTTP URL of the local fine-tuned model server.
+            When None and backend is "auto" or "fine_tuned", this backend
+            is skipped.
+
+    Returns:
+        A KushyaResult. The kushya field may be None if no backend produced
+        a substantive challenge — the caller must handle that case
+        (typically by treating the interpretation as un-challenged).
+    """
+    if backend == "fine_tuned":
+        return _try_fine_tuned(interpretation, fine_tuned_endpoint, allow_fallback=False)
+    if backend == "external":
+        return _try_external(interpretation, allow_fallback=False)
+    if backend == "fallback":
+        return _fallback(interpretation)
+
+    # auto: try in priority order
+    if fine_tuned_endpoint:
+        result = _try_fine_tuned(interpretation, fine_tuned_endpoint, allow_fallback=True)
+        if result.kushya:
+            return result
+
+    result = _try_external(interpretation, allow_fallback=True)
+    if result.kushya:
+        return result
+
+    return _fallback(interpretation)
+
+
+def _try_fine_tuned(
+    interpretation: str,
+    endpoint: Optional[str],
+    *,
+    allow_fallback: bool,
+) -> KushyaResult:
+    """Call the fine-tuned local kushya model over HTTP.
+
+    The serve.py script wraps llama.cpp behind a small HTTP API at
+    POST /generate that accepts {"prompt": "..."} and returns
+    {"completion": "..."}.
+    """
+    if not endpoint:
+        if allow_fallback:
+            return KushyaResult(
+                interpretation=interpretation,
+                kushya=None,
+                backend="fine_tuned",
+                torah_refs=[],
+                notes="fine-tuned endpoint not configured",
+            )
+        raise ValueError("fine_tuned backend requires fine_tuned_endpoint")
+
+    try:
+        import httpx
+    except ImportError:
+        return KushyaResult(
+            interpretation=interpretation,
+            kushya=None,
+            backend="fine_tuned",
+            torah_refs=[],
+            notes="httpx not installed",
+        )
+
+    prompt = (
+        "Generate a kushya (challenge) for this interpretation. "
+        "Cite a specific Torah passage. End with a question mark.\n\n"
+        f"Proposed interpretation: {interpretation}\n\n"
+        "Kushya: "
+    )
+
+    try:
+        with httpx.Client(timeout=60.0) as client:
+            response = client.post(endpoint, json={"prompt": prompt})
+            response.raise_for_status()
+            completion = response.json().get("completion", "").strip()
+    except httpx.RequestError as e:
+        if allow_fallback:
+            return KushyaResult(
+                interpretation=interpretation,
+                kushya=None,
+                backend="fine_tuned",
+                torah_refs=[],
+                notes=f"endpoint unreachable: {e}",
+            )
+        raise
+
+    if not completion:
+        return KushyaResult(
+            interpretation=interpretation,
+            kushya=None,
+            backend="fine_tuned",
+            torah_refs=[],
+            raw_response=completion,
+            notes="empty completion",
+        )
+
+    return KushyaResult(
+        interpretation=interpretation,
+        kushya=completion,
+        backend="fine_tuned",
+        torah_refs=_extract_refs(completion),
+        raw_response=completion,
+    )
+
+
+def _try_external(interpretation: str, *, allow_fallback: bool) -> KushyaResult:
+    """Call an external model (Claude API) via Korban Processor.
+
+    The output flows through the offering protocol so external response is
+    treated as testimony, not law (eish_zarah_reading.md). Provenance is
+    preserved on the resulting consultation record.
+    """
+    if consult_external_model is None:
+        if allow_fallback:
+            return KushyaResult(
+                interpretation=interpretation,
+                kushya=None,
+                backend="external",
+                torah_refs=[],
+                notes="korban_processor not on sys.path",
+            )
+        raise RuntimeError("korban_processor not on sys.path")
+
+    system_prompt = (
+        "You are a Talmudic challenger. Generate exactly one kushya (challenge) "
+        "to the proposed interpretation. The kushya must:\n"
+        "  1. Reference a specific Torah passage that creates tension\n"
+        "  2. Identify the logical or textual weakness\n"
+        "  3. Be expressed as a question, not a statement\n"
+        "  4. Not be resolvable by trivial clarification\n\n"
+        "Return only the kushya question. No preamble, no closing."
+    )
+    query = f"Proposed interpretation: {interpretation}"
+
+    try:
+        consultation = consult_external_model(
+            query=query,
+            source_query=f"kushya_request:{interpretation[:80]}",
+            provenance=Provenance.CLAUDE,
+            offering_type=OfferingType.OLAH,
+            system_prompt=system_prompt,
+        )
+    except Exception as e:
+        if allow_fallback:
+            return KushyaResult(
+                interpretation=interpretation,
+                kushya=None,
+                backend="external",
+                torah_refs=[],
+                notes=f"external consultation failed: {type(e).__name__}: {e}",
+            )
+        raise
+
+    raw = consultation.raw_response.content
+    vital = consultation.processed.vital_content
+
+    # The vital content already passed through shechitah (preamble strip)
+    # and zerikah (vital extraction). Use it as the kushya.
+    if not vital.strip():
+        return KushyaResult(
+            interpretation=interpretation,
+            kushya=None,
+            backend="external",
+            torah_refs=[],
+            raw_response=raw,
+            notes="zerikah found no vital content in external response",
+        )
+
+    return KushyaResult(
+        interpretation=interpretation,
+        kushya=vital,
+        backend="external",
+        torah_refs=_extract_refs(vital),
+        raw_response=raw,
+        notes=(
+            "External testimony — recorded under offering protocol. "
+            "NOT binding ruling."
+        ),
+    )
+
+
+def _fallback(interpretation: str) -> KushyaResult:
+    """No external backend available. Return a structured no-kushya result.
+
+    The Talmudic Engine has its own rule-based kushya generator
+    (challenge.pl). This fallback is used when even external testimony
+    is unavailable — we do NOT invent a kushya, we honestly report
+    that no challenge was generated.
+    """
+    return KushyaResult(
+        interpretation=interpretation,
+        kushya=None,
+        backend="fallback",
+        torah_refs=[],
+        notes=(
+            "No backend available. Rule-based kushyot from the Talmudic "
+            "Engine should be used; the system does not invent challenges."
+        ),
+    )
+
+
+def _extract_refs(text: str) -> list[str]:
+    """Extract Torah verse references from kushya text.
+
+    Returns the full match strings (e.g. "Shemot 35:3", "Devarim 21:1-9"),
+    not just the captured book names.
+    """
+    import re
+    # Non-capturing group so re.findall returns the full match.
+    pattern = r"\b(?:Bereishit|Shemot|Vayikra|Bamidbar|Devarim)\s+\d+:\d+(?:[-–]\d+)?"
+    return re.findall(pattern, text)

kushya_generator/test/test_dataset.py

@@ -0,0 +1,147 @@
+"""
+Validate the kushya training dataset structure.
+
+Architecture §3.6 — every training pair must satisfy the four requirements
+or be rejected. These tests run prepare_training.py's validator against
+the seed pairs to catch dataset regressions.
+"""
+
+import json
+import sys
+from pathlib import Path
+
+# Add parent data/ dir to path so we can import prepare_training
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "data"))
+
+from prepare_training import (  # noqa: E402
+    REQUIRED_TYPES,
+    ValidationError,
+    load_jsonl,
+    to_training_format,
+    validate_pair,
+)
+
+
+SEED_PATH = (
+    Path(__file__).resolve().parent.parent / "data" / "training_pairs" / "seed_pairs.jsonl"
+)
+
+
+def test_seed_pairs_file_exists():
+    """The seed pairs file must exist — it bootstraps the training pipeline."""
+    assert SEED_PATH.exists(), f"seed_pairs.jsonl not found at {SEED_PATH}"
+
+
+def test_seed_pairs_validate():
+    """Every seed pair must pass validation. If a pair fails, the dataset
+    has regressed and the training pipeline would silently degrade."""
+    failures = []
+    for line_no, pair in load_jsonl(SEED_PATH):
+        try:
+            validate_pair(pair, "seed_pairs.jsonl", line_no)
+        except ValidationError as e:
+            failures.append(str(e))
+    assert not failures, "seed_pairs.jsonl has invalid records:\n" + "\n".join(failures)
+
+
+def test_seed_pairs_cover_all_kushya_types():
+    """The seed set should exercise every kushya type, so the training
+    split is stratified across the categories."""
+    types_seen = set()
+    for _, pair in load_jsonl(SEED_PATH):
+        types_seen.add(pair["type"])
+
+    # We don't require ALL types to be in the small seed (10 pairs),
+    # but we do require at least 3 types to ensure stratification works.
+    assert len(types_seen) >= 3, (
+        f"Only {len(types_seen)} types in seed: {types_seen}. "
+        f"Need at least 3 for stratified split."
+    )
+
+    # Every type used must be a known type (validator already checks this,
+    # but we want a clear failure if a typo creeps in).
+    unknown = types_seen - REQUIRED_TYPES
+    assert not unknown, f"Unknown types in seed: {unknown}"
+
+
+def test_to_training_format_produces_expected_shape():
+    """The training format must follow Architecture §3.6:
+      input:  'Proposed interpretation: ...'
+      output: 'Kushya: ...'
+    """
+    sample = {
+        "interpretation": "Cooking on Shabbat is forbidden.",
+        "kushya": "What about cooking by direct sunlight where no kindling occurs in Shemot 35:3?",
+        "torah_refs": ["Shemot 35:3"],
+        "type": "scope_narrowing",
+    }
+    formatted = to_training_format(sample)
+    assert formatted["input"].startswith("Proposed interpretation:")
+    assert formatted["output"].startswith("Kushya:")
+    assert formatted["metadata"]["type"] == "scope_narrowing"
+    assert formatted["metadata"]["torah_refs"] == ["Shemot 35:3"]
+
+
+# ============================================================
+# Validator regression tests — these should REJECT bad pairs.
+# ============================================================
+
+def _expect_rejection(pair, must_contain):
+    try:
+        validate_pair(pair, "test", 1)
+    except ValidationError as e:
+        assert must_contain in str(e), (
+            f"Expected error containing {must_contain!r}, got: {e}"
+        )
+    else:
+        raise AssertionError(f"Pair should have been rejected: {pair}")
+
+
+def test_validator_rejects_empty_kushya():
+    pair = {
+        "interpretation": "Some interp.",
+        "kushya": "",
+        "torah_refs": ["Shemot 21:1"],
+        "type": "internal_tension",
+    }
+    _expect_rejection(pair, "kushya must be a non-empty string")
+
+
+def test_validator_rejects_kushya_without_question_form():
+    pair = {
+        "interpretation": "Some interp.",
+        "kushya": "There is a problem with this interpretation that needs clarification.",
+        "torah_refs": ["Shemot 21:1"],
+        "type": "internal_tension",
+    }
+    _expect_rejection(pair, "expressible as a question")
+
+
+def test_validator_rejects_too_short_kushya():
+    pair = {
+        "interpretation": "Some interp.",
+        "kushya": "Why?",
+        "torah_refs": ["Shemot 21:1"],
+        "type": "internal_tension",
+    }
+    _expect_rejection(pair, "too short")
+
+
+def test_validator_rejects_no_torah_reference():
+    pair = {
+        "interpretation": "Some interp.",
+        "kushya": "What does this even mean in context — there is internal tension here is there not?",
+        "torah_refs": [],
+        "type": "internal_tension",
+    }
+    _expect_rejection(pair, "must cite a specific Torah passage")
+
+
+def test_validator_rejects_unknown_type():
+    pair = {
+        "interpretation": "Some interp.",
+        "kushya": "What does this mean in Shemot 21:1, given the apparent tension?",
+        "torah_refs": ["Shemot 21:1"],
+        "type": "made_up_type",
+    }
+    _expect_rejection(pair, "not in known types")

kushya_generator/test/test_kushya_api.py

@@ -0,0 +1,124 @@
+"""
+Tests for the kushya generator API.
+
+These tests use mocked external callers; no real API requests are made.
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+# Add parent inference/ dir
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "inference"))
+# Add the shorashim repo's korban_processor src to sys.path
+sys.path.insert(
+    0,
+    str(Path(__file__).resolve().parent.parent.parent / "korban_processor" / "src"),
+)
+
+from kushya_api import KushyaResult, generate_kushya, _extract_refs  # noqa: E402
+
+
+# ============================================================
+# Backend selection
+# ============================================================
+
+def test_fallback_returns_no_kushya_with_explanation():
+    """The fallback path honestly reports no challenge — never invents one."""
+    result = generate_kushya("Some interpretation.", backend="fallback")
+    assert isinstance(result, KushyaResult)
+    assert result.kushya is None
+    assert result.backend == "fallback"
+    assert "does not invent" in (result.notes or "")
+
+
+def test_fine_tuned_without_endpoint_falls_back_in_auto():
+    """auto mode skips fine_tuned when no endpoint is configured, then
+    skips external when not available, ending in fallback."""
+    # We expect the Korban Processor's external model call to either succeed
+    # (if API key is present) or fail (if not) — either way, the result has
+    # a clear backend. Mock to ensure deterministic behavior.
+    from korban_processor.external_offering import consult_external_model  # noqa: F401
+
+    with patch("kushya_api.consult_external_model") as mock_consult:
+        mock_consult.side_effect = RuntimeError("simulated API failure")
+        result = generate_kushya("Some interpretation.", backend="auto")
+
+    assert result.kushya is None
+    assert result.backend in ("fallback", "external")
+
+
+def test_fine_tuned_explicit_without_endpoint_raises():
+    """When backend='fine_tuned' is forced but no endpoint is configured,
+    we raise rather than silently falling through."""
+    import pytest
+    with pytest.raises(ValueError, match="fine_tuned_endpoint"):
+        generate_kushya("Some interpretation.", backend="fine_tuned")
+
+
+# ============================================================
+# External backend — mocked, exercises the full eish-zarah path
+# ============================================================
+
+def test_external_backend_returns_kushya_via_korban_processor():
+    """When external consultation succeeds, the result is recorded with
+    'external' backend and a notes field that emphasizes it is testimony,
+    not ruling."""
+    from kushya_api import _try_external
+
+    fake_consultation = type("FakeConsultation", (), {})()
+    fake_consultation.raw_response = type(
+        "FakeResponse", (), {"content": "raw response with a question?"}
+    )()
+    fake_consultation.processed = type(
+        "FakeProcessed",
+        (),
+        {"vital_content": "What about Shemot 35:3 — does the prohibition extend to passive heating?"},
+    )()
+
+    with patch("kushya_api.consult_external_model", return_value=fake_consultation):
+        result = _try_external("Cooking on Shabbat is forbidden.", allow_fallback=True)
+
+    assert result.kushya
+    assert result.backend == "external"
+    assert "testimony" in (result.notes or "").lower()
+    assert "Shemot 35:3" in result.torah_refs
+
+
+def test_external_backend_handles_empty_vital_content():
+    """If zerikah extracts nothing vital from external response, we
+    record None — we do not pretend an empty kushya is real."""
+    from kushya_api import _try_external
+
+    fake_consultation = type("FakeConsultation", (), {})()
+    fake_consultation.raw_response = type("FakeResp", (), {"content": "Great question!"})()
+    fake_consultation.processed = type("FakeProc", (), {"vital_content": ""})()
+
+    with patch("kushya_api.consult_external_model", return_value=fake_consultation):
+        result = _try_external("Some interp.", allow_fallback=True)
+
+    assert result.kushya is None
+    assert result.backend == "external"
+    assert "no vital content" in (result.notes or "").lower()
+
+
+# ============================================================
+# Reference extraction
+# ============================================================
+
+def test_extract_refs_finds_torah_citations():
+    text = "What about Shemot 35:3 and also Vayikra 18:5?"
+    refs = _extract_refs(text)
+    assert "Shemot 35:3" in refs
+    assert "Vayikra 18:5" in refs
+
+
+def test_extract_refs_handles_verse_ranges():
+    text = "Devarim 21:1-9 and Bereishit 1:1"
+    refs = _extract_refs(text)
+    assert any("Devarim 21:1" in r for r in refs)
+    assert "Bereishit 1:1" in refs
+
+
+def test_extract_refs_returns_empty_for_no_citations():
+    assert _extract_refs("This text has no Torah references.") == []