Training phase: scripts 09+10, --model flag in 03, gold schema patch

smoke_signal/scripts/03_ocr_bakeoff.py

@@ -98,18 +98,56 @@ def load_page_profile(book_id: str) -> Optional[dict]:
 
 
 # ── Surya OCR ──────────────────────────────────────────────────────────────────
-def _run_surya(image_path: Path, langs: list) -> dict:
+def _safe_load_surya_component(loader, checkpoint: Optional[str]):
     """
-    Run Surya OCR on a single page image.
-    Returns standardised result dict.
+    Attempt to pass checkpoint to Surya loader, with safe fallback for older APIs.
+    """
+    if not checkpoint:
+        return loader()
+    try:
+        return loader(checkpoint=checkpoint)
+    except TypeError:
+        return loader()
+
+
+def load_surya_context(checkpoint: Optional[str] = None) -> Optional[dict]:
+    """
+    Load Surya OCR models once per run.
+    Returns context dict or None if Surya import/loading fails.
     """
     try:
-        from PIL import Image
         from surya.ocr import run_ocr
         from surya.model.detection.model import load_model as load_det_model
         from surya.model.detection.processor import load_processor as load_det_processor
         from surya.model.recognition.model import load_model as load_rec_model
         from surya.model.recognition.processor import load_processor as load_rec_processor
+    except ImportError:
+        return None
+
+    try:
+        det_model = _safe_load_surya_component(load_det_model, checkpoint)
+        det_processor = _safe_load_surya_component(load_det_processor, checkpoint)
+        rec_model = _safe_load_surya_component(load_rec_model, checkpoint)
+        rec_processor = _safe_load_surya_component(load_rec_processor, checkpoint)
+        return {
+            "run": run_ocr,
+            "det_model": det_model,
+            "det_processor": det_processor,
+            "rec_model": rec_model,
+            "rec_processor": rec_processor,
+            "checkpoint": checkpoint,
+        }
+    except Exception:
+        return None
+
+
+def _run_surya(image_path: Path, langs: list, surya_ctx: Optional[dict] = None) -> dict:
+    """
+    Run Surya OCR on a single page image.
+    Returns standardised result dict.
+    """
+    try:
+        from PIL import Image
     except ImportError as e:
         return {
             "engine": "surya",
@@ -119,21 +157,25 @@ def _run_surya(image_path: Path, langs: list) -> dict:
             "confidence": 0.0,
         }
 
+    ctx = surya_ctx or load_surya_context()
+    if not ctx:
+        return {
+            "engine": "surya",
+            "error":  "Surya model load failed. Check surya-ocr install and checkpoint path.",
+            "text":   "",
+            "words":  [],
+            "confidence": 0.0,
+        }
+
     try:
         image = Image.open(str(image_path)).convert("RGB")
-
-        det_model     = load_det_model()
-        det_processor = load_det_processor()
-        rec_model     = load_rec_model()
-        rec_processor = load_rec_processor()
-
-        results = run_ocr(
+        results = ctx["run"](
             [image],
             [langs],
-            det_model,
-            det_processor,
-            rec_model,
-            rec_processor,
+            ctx["det_model"],
+            ctx["det_processor"],
+            ctx["rec_model"],
+            ctx["rec_processor"],
         )
 
         page_result = results[0]
@@ -163,6 +205,7 @@ def _run_surya(image_path: Path, langs: list) -> dict:
             "words":      words,
             "confidence": avg_conf,
             "line_count": len(words),
+            "model_checkpoint": ctx.get("checkpoint") or "base",
             "error":      None,
         }
 
@@ -173,6 +216,7 @@ def _run_surya(image_path: Path, langs: list) -> dict:
             "text":       "",
             "words":      [],
             "confidence": 0.0,
+            "model_checkpoint": ctx.get("checkpoint") or "base",
         }
 
 
@@ -269,6 +313,7 @@ def ocr_page(
     page_num: int,
     engine: str,
     config: dict,
+    surya_ctx: Optional[dict] = None,
     dry_run: bool = False,
 ) -> dict:
     """Run OCR on one page, save result, return summary."""
@@ -290,7 +335,7 @@ def ocr_page(
         return result
 
     if engine == "surya":
-        ocr_out = _run_surya(image_path, config["surya_langs"])
+        ocr_out = _run_surya(image_path, config["surya_langs"], surya_ctx=surya_ctx)
     elif engine == "tesseract":
         ocr_out = _run_tesseract(image_path, config["tesseract_lang"], config["tesseract_psm"])
     else:
@@ -310,7 +355,13 @@ def ocr_page(
 
 
 # ── Per-book OCR runner ────────────────────────────────────────────────────────
-def ocr_book(record: dict, engine: str, dry_run: bool = False) -> dict:
+def ocr_book(
+    record: dict,
+    engine: str,
+    config: dict,
+    surya_ctx: Optional[dict] = None,
+    dry_run: bool = False,
+) -> dict:
     book_id  = record["book_id"]
     print(f"\n  [{book_id}] {record['filename']} — engine: {engine}")
 
@@ -319,7 +370,7 @@ def ocr_book(record: dict, engine: str, dry_run: bool = False) -> dict:
         print(f"    ✗ No page profile found. Run 02_profile_pdfs.py first.")
         return {"book_id": book_id, "error": "no_profile", "pages": []}
 
-    eligible_routes = OCR_CONFIG["eligible_routes"]
+    eligible_routes = config["eligible_routes"]
     ocr_pages = [p for p in profile["pages"] if p.get("route") in eligible_routes]
 
     print(f"    OCR-eligible pages: {len(ocr_pages)} / {profile['page_count']}")
@@ -356,7 +407,7 @@ def ocr_book(record: dict, engine: str, dry_run: bool = False) -> dict:
             errors.append({"page": page_num, "error": "render_missing"})
             continue
 
-        result = ocr_page(image_path, book_id, page_num, engine, OCR_CONFIG, dry_run)
+        result = ocr_page(image_path, book_id, page_num, engine, config, surya_ctx=surya_ctx, dry_run=dry_run)
         page_results.append(result)
 
         conf = result.get("confidence", 0.0)
@@ -407,18 +458,28 @@ def main():
     parser.add_argument("--batch-id", help="Tag this run with a batch ID")
     parser.add_argument("--engine",   choices=["surya", "tesseract", "both"],
                         default="surya", help="OCR engine to use (default: surya)")
+    parser.add_argument("--model",    default=None,
+                        help="Optional Surya checkpoint/model path for OCR engine=surya")
     parser.add_argument("--dry-run",  action="store_true", help="No files written")
     parser.add_argument("--all",      action="store_true", help="Include already-OCRed books")
     args = parser.parse_args()
 
     run_id  = args.batch_id or f"SS-RUN-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}"
     engines = ["surya", "tesseract"] if args.engine == "both" else [args.engine]
+    run_config = dict(OCR_CONFIG)
+    if args.model:
+        run_config["surya_model_checkpoint"] = args.model
+        # Governance: checkpoint changes require a new config version.
+        model_tag = hashlib.sha256(args.model.encode("utf-8")).hexdigest()[:8]
+        run_config["config_version"] = f"{OCR_CONFIG['config_version']}_ft_{model_tag}"
 
     print(f"\n{'='*60}")
     print(f"  Smoke Signal — Stage 4: OCR Bake-Off")
     print(f"  Run ID  : {run_id}")
     print(f"  Engines : {engines}")
-    print(f"  Config  : {OCR_CONFIG['config_version']}")
+    print(f"  Config  : {run_config['config_version']}")
+    if args.model:
+        print(f"  Surya model override : {args.model}")
     if args.dry_run:
         print(f"  Mode    : DRY RUN")
     print(f"{'='*60}")
@@ -439,19 +500,30 @@ def main():
     else:
         books = [r for r in manifest.values() if r.get("status") in eligible_statuses]
 
+    # Governance: never process unknown/excluded rights in OCR batches.
+    books = [r for r in books if r.get("rights_class") not in ("unknown", "excluded")]
+
     if not books:
-        print(f"\n  No books with status in {eligible_statuses}.")
-        print("  Run 02_profile_pdfs.py first to render pages.")
+        print(f"\n  No books eligible after status/rights filters.")
+        print(f"  Eligible statuses: {eligible_statuses}")
+        print("  Rights blocked: unknown, excluded")
         sys.exit(0)
 
     print(f"\n  Books to OCR: {len(books)}")
 
     all_results = []
     t_start     = time.time()
+    surya_ctx = None
+    if "surya" in engines and not args.dry_run:
+        surya_ctx = load_surya_context(args.model)
+        if not surya_ctx:
+            print("\n  [error] Could not load Surya models/checkpoint.")
+            print("  Check surya-ocr install and --model path.")
+            sys.exit(1)
 
     for record in books:
         for engine in engines:
-            result = ocr_book(record, engine, dry_run=args.dry_run)
+            result = ocr_book(record, engine, run_config, surya_ctx=surya_ctx, dry_run=args.dry_run)
             all_results.append(result)
 
             # Update manifest status
@@ -461,7 +533,7 @@ def main():
     # Save manifest + config + run log
     if not args.dry_run:
         save_manifest(manifest)
-        save_ocr_config(OCR_CONFIG)
+        save_ocr_config(run_config)
 
         log_path = LOGS_DIR / f"{run_id}_ocr_bakeoff.json"
         with open(log_path, "w", encoding="utf-8") as f:
@@ -469,7 +541,8 @@ def main():
                 "run_id":   run_id,
                 "run_at":   datetime.utcnow().isoformat() + "Z",
                 "engines":  engines,
-                "config":   OCR_CONFIG,
+                "config":   run_config,
+                "surya_model_checkpoint": args.model or "base",
                 "results":  all_results,
             }, f, indent=2)
         print(f"\n  Run log → {log_path.relative_to(ROOT)}")

smoke_signal/scripts/09_finetune_surya.py

@@ -0,0 +1,620 @@
+#!/usr/bin/env python3
+"""
+Smoke Signal — Stage 11: Surya Fine-Tuning Orchestrator
+========================================================
+Builds a governance-safe OCR fine-tuning dataset from gold corrections,
+optionally uploads it to Hugging Face, and can launch Surya OCR finetuning.
+
+Governance controls enforced:
+  - unknown/excluded rights are always blocked
+  - mixed rights classes are blocked by default
+  - checkpoint/config changes are versioned and logged
+  - every run logs model version, dataset size, and gold set hash
+
+Primary sources used for integration choices:
+  - Surya README finetune entrypoint and args
+  - Surya example dataset shape (`image` + `text`)
+"""
+
+import argparse
+import csv
+import hashlib
+import json
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+ROOT = Path(__file__).resolve().parents[1]
+MANIFEST_CSV = ROOT / "manifest" / "source_manifest.csv"
+RUN_LOG_CSV = ROOT / "manifest" / "run_log.csv"
+GOLD_FILE = ROOT / "gold" / "gold_corrections.jsonl"
+RENDERS_DIR = ROOT / "renders"
+TRAINING_DIR = ROOT / "training"
+TRAINING_DATASETS_DIR = TRAINING_DIR / "datasets"
+TRAINING_RUNS_DIR = TRAINING_DIR / "runs"
+
+ELIGIBLE_RIGHTS = {"public-domain", "licensed-owned", "controlled-internal"}
+BLOCKED_RIGHTS = {"unknown", "excluded"}
+
+RUN_LOG_FIELDS = [
+    "run_id",
+    "date",
+    "operator",
+    "config_version",
+    "schema_version",
+    "source_batch",
+    "pages_processed",
+    "errors",
+    "cost_usd",
+    "output_path",
+    "notes",
+]
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def utc_iso() -> str:
+    return utc_now().isoformat().replace("+00:00", "Z")
+
+
+def ensure_dirs() -> None:
+    TRAINING_DIR.mkdir(parents=True, exist_ok=True)
+    TRAINING_DATASETS_DIR.mkdir(parents=True, exist_ok=True)
+    TRAINING_RUNS_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def sha256_file(path: Path) -> str:
+    h = hashlib.sha256()
+    with open(path, "rb") as f:
+        for block in iter(lambda: f.read(1 << 20), b""):
+            h.update(block)
+    return h.hexdigest()
+
+
+def sha256_text(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def ensure_run_log() -> None:
+    RUN_LOG_CSV.parent.mkdir(parents=True, exist_ok=True)
+    if RUN_LOG_CSV.exists():
+        return
+    with open(RUN_LOG_CSV, "w", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=RUN_LOG_FIELDS)
+        writer.writeheader()
+
+
+def append_run_log(row: Dict[str, str]) -> None:
+    ensure_run_log()
+    with open(RUN_LOG_CSV, "a", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=RUN_LOG_FIELDS)
+        writer.writerow({k: row.get(k, "") for k in RUN_LOG_FIELDS})
+
+
+def load_manifest() -> Dict[str, Dict[str, str]]:
+    records: Dict[str, Dict[str, str]] = {}
+    if not MANIFEST_CSV.exists():
+        return records
+
+    with open(MANIFEST_CSV, newline="", encoding="utf-8") as f:
+        for row in csv.DictReader(f):
+            book_id = str(row.get("book_id", "")).strip()
+            if book_id:
+                records[book_id] = row
+    return records
+
+
+def load_gold_records(path: Path) -> List[Dict]:
+    records: List[Dict] = []
+    if not path.exists():
+        return records
+
+    with open(path, encoding="utf-8") as f:
+        for idx, line in enumerate(f, start=1):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                rec = json.loads(line)
+                rec["_gold_line"] = idx
+                records.append(rec)
+            except json.JSONDecodeError:
+                # Keep pipeline resilient: skip malformed line.
+                continue
+    return records
+
+
+def parse_page_number(value) -> Optional[int]:
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def resolve_image_path(record: Dict, book_id: str, page_num: Optional[int]) -> Optional[Path]:
+    path_fields = [
+        "page_image_path",
+        "page_image",
+        "image_path",
+        "crop_path",
+        "render_path",
+    ]
+    for field in path_fields:
+        raw = record.get(field)
+        if not raw:
+            continue
+        candidate = Path(str(raw))
+        if not candidate.is_absolute():
+            candidate = ROOT / candidate
+        if candidate.exists() and candidate.is_file():
+            return candidate.resolve()
+
+    if page_num is not None:
+        render_dir = RENDERS_DIR / book_id
+        if render_dir.exists():
+            candidates = sorted(render_dir.glob(f"{book_id}_page_{page_num:04d}_*.*"))
+            if candidates:
+                return candidates[0].resolve()
+
+    return None
+
+
+def select_transcript(record: Dict) -> str:
+    for key in ("final_text", "corrected_text", "text", "raw_text", "raw_ocr"):
+        value = record.get(key)
+        if value is not None:
+            text = str(value).strip()
+            if text:
+                return text
+    return ""
+
+
+def build_examples(
+    gold_records: List[Dict],
+    manifest: Dict[str, Dict[str, str]],
+    rights_class_filter: Optional[str],
+    allow_mixed_rights: bool,
+) -> Tuple[List[Dict], Dict]:
+    stats = {
+        "gold_rows": len(gold_records),
+        "kept": 0,
+        "skipped_missing_book": 0,
+        "skipped_missing_manifest": 0,
+        "skipped_status": 0,
+        "skipped_blocked_rights": 0,
+        "skipped_rights_filter": 0,
+        "skipped_missing_text": 0,
+        "skipped_missing_image": 0,
+    }
+
+    rights_seen = set()
+    examples: List[Dict] = []
+
+    for rec in gold_records:
+        book_id = str(rec.get("book_id", "")).strip()
+        if not book_id:
+            stats["skipped_missing_book"] += 1
+            continue
+
+        manifest_row = manifest.get(book_id)
+        if not manifest_row:
+            stats["skipped_missing_manifest"] += 1
+            continue
+
+        status = str(rec.get("status", "")).strip().lower()
+        if status and status not in {"accepted", "edited"}:
+            stats["skipped_status"] += 1
+            continue
+
+        rights_class = str(manifest_row.get("rights_class", "unknown")).strip().lower()
+        if rights_class in BLOCKED_RIGHTS or rights_class not in ELIGIBLE_RIGHTS:
+            stats["skipped_blocked_rights"] += 1
+            continue
+
+        if rights_class_filter and rights_class != rights_class_filter:
+            stats["skipped_rights_filter"] += 1
+            continue
+
+        text = select_transcript(rec)
+        if not text:
+            stats["skipped_missing_text"] += 1
+            continue
+
+        page_num = parse_page_number(rec.get("page") or rec.get("page_number"))
+        image_path = resolve_image_path(rec, book_id, page_num)
+        if not image_path:
+            stats["skipped_missing_image"] += 1
+            continue
+
+        rights_seen.add(rights_class)
+
+        example = {
+            "image": str(image_path),
+            "text": text,
+            "book_id": book_id,
+            "page": page_num,
+            "region_class": str(rec.get("region_class", "narration")),
+            "rights_class": rights_class,
+            "confidence": float(rec.get("confidence", 0) or 0),
+            "gold_line": int(rec.get("_gold_line", 0)),
+        }
+        examples.append(example)
+
+    if not allow_mixed_rights and len(rights_seen) > 1:
+        raise ValueError(
+            f"Mixed rights classes found in training set: {sorted(rights_seen)}. "
+            "Run separate jobs per rights class or pass --allow-mixed-rights explicitly."
+        )
+
+    stats["kept"] = len(examples)
+    stats["rights_seen"] = sorted(rights_seen)
+    return examples, stats
+
+
+def save_training_artifacts(run_id: str, examples: List[Dict], stats: Dict, metadata: Dict) -> Path:
+    run_dataset_dir = TRAINING_DATASETS_DIR / run_id
+    run_dataset_dir.mkdir(parents=True, exist_ok=True)
+
+    examples_jsonl = run_dataset_dir / "training_examples.jsonl"
+    with open(examples_jsonl, "w", encoding="utf-8") as f:
+        for row in examples:
+            f.write(json.dumps(row, ensure_ascii=False) + "\n")
+
+    manifest_path = run_dataset_dir / "dataset_manifest.json"
+    with open(manifest_path, "w", encoding="utf-8") as f:
+        json.dump({"stats": stats, "metadata": metadata}, f, indent=2, ensure_ascii=False)
+
+    return run_dataset_dir
+
+
+def build_hf_dataset(examples: List[Dict]):
+    try:
+        from datasets import Dataset, Image
+    except Exception as exc:  # pragma: no cover - environment-dependent
+        raise RuntimeError(
+            "datasets[vision] is required. Install with: pip install datasets[vision]"
+        ) from exc
+
+    ds = Dataset.from_dict({
+        "image": [e["image"] for e in examples],
+        "text": [e["text"] for e in examples],
+    }).cast_column("image", Image())
+    return ds
+
+
+def push_dataset_to_hub(ds, repo_id: str, private: bool, token: Optional[str], run_id: str) -> None:
+    ds.push_to_hub(
+        repo_id,
+        private=private,
+        token=token,
+        commit_message=f"Smoke Signal finetune dataset {run_id}",
+    )
+
+
+def resolve_finetune_entrypoint(explicit_script: Optional[str], surya_repo: Optional[str]) -> List[str]:
+    if explicit_script:
+        script_path = Path(explicit_script).expanduser().resolve()
+        if not script_path.exists():
+            raise FileNotFoundError(f"Surya finetune script not found: {script_path}")
+        return [sys.executable, str(script_path)]
+
+    try:
+        import importlib.util
+
+        spec = importlib.util.find_spec("surya.scripts.finetune_ocr")
+        if spec is not None:
+            return [sys.executable, "-m", "surya.scripts.finetune_ocr"]
+    except Exception:
+        pass
+
+    if surya_repo:
+        candidate = Path(surya_repo).expanduser().resolve() / "surya" / "scripts" / "finetune_ocr.py"
+        if candidate.exists():
+            return [sys.executable, str(candidate)]
+
+    raise RuntimeError(
+        "Could not resolve Surya finetune entrypoint. "
+        "Install surya-ocr or pass --surya-finetune-script /path/to/finetune_ocr.py"
+    )
+
+
+def _detect_hub_model_arg_name(entrypoint_cmd: List[str]) -> str:
+    """
+    TrainingArguments changed over time. Detect supported hub model id arg
+    from finetune --help output.
+    """
+    try:
+        probe = subprocess.run(
+            entrypoint_cmd + ["--help"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        help_text = (probe.stdout or "") + "\\n" + (probe.stderr or "")
+        if "--hub_model_id" in help_text:
+            return "--hub_model_id"
+        if "--push_to_hub_model_id" in help_text:
+            return "--push_to_hub_model_id"
+    except Exception:
+        pass
+    # Default to current TrainingArguments key.
+    return "--hub_model_id"
+
+
+def build_train_command(args, run_output_dir: Path) -> List[str]:
+    cmd = resolve_finetune_entrypoint(args.surya_finetune_script, args.surya_repo)
+    hub_model_arg = _detect_hub_model_arg_name(cmd)
+    cmd += [
+        "--output_dir",
+        str(run_output_dir),
+        "--dataset_name",
+        args.dataset_repo_id,
+        "--per_device_train_batch_size",
+        str(args.per_device_train_batch_size),
+        "--gradient_checkpointing",
+        "true" if args.gradient_checkpointing else "false",
+        "--max_sequence_length",
+        str(args.max_sequence_length),
+        "--num_train_epochs",
+        str(args.num_train_epochs),
+        "--learning_rate",
+        str(args.learning_rate),
+        "--logging_steps",
+        str(args.logging_steps),
+        "--save_steps",
+        str(args.save_steps),
+        "--save_total_limit",
+        str(args.save_total_limit),
+        "--remove_unused_columns",
+        "false",
+        "--push_to_hub",
+        "true",
+        hub_model_arg,
+        args.model_repo_id,
+    ]
+
+    if args.pretrained_checkpoint_path:
+        cmd += ["--pretrained_checkpoint_path", args.pretrained_checkpoint_path]
+
+    if args.hf_token:
+        cmd += ["--hub_token", args.hf_token]
+
+    return cmd
+
+
+def write_run_summary(path: Path, summary: Dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(summary, f, indent=2, ensure_ascii=False)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Smoke Signal — Stage 11: Surya Fine-Tuning")
+    parser.add_argument("--gold-file", default=str(GOLD_FILE), help="Path to gold corrections JSONL")
+    parser.add_argument(
+        "--rights-class",
+        default=None,
+        choices=sorted(ELIGIBLE_RIGHTS),
+        help="Restrict to one rights class (recommended governance mode)",
+    )
+    parser.add_argument(
+        "--allow-mixed-rights",
+        action="store_true",
+        help="Allow mixed rights classes in one run (off by default)",
+    )
+    parser.add_argument(
+        "--dataset-repo-id",
+        default="Pointf5ive/smoke-signal-ocr-finetune",
+        help="HF dataset repo id (<namespace>/<name>)",
+    )
+    parser.add_argument(
+        "--model-repo-id",
+        default="Pointf5ive/smoke-signal-surya-ft",
+        help="HF model repo id (<namespace>/<name>)",
+    )
+    parser.add_argument("--hf-token", default=None, help="HF token (or set HF_TOKEN env var)")
+    parser.add_argument("--private-dataset", action="store_true", help="Create/push dataset repo as private")
+    parser.add_argument("--private-model", action="store_true", help="Create model repo as private")
+    parser.add_argument("--operator", default="codex", help="Operator name for governance logs")
+
+    parser.add_argument("--pretrained-checkpoint-path", default=None, help="Optional Surya init checkpoint")
+    parser.add_argument("--surya-finetune-script", default=None, help="Path to surya/scripts/finetune_ocr.py")
+    parser.add_argument("--surya-repo", default=None, help="Path to local Surya repo (fallback resolver)")
+
+    parser.add_argument("--per-device-train-batch-size", type=int, default=16)
+    parser.add_argument("--max-sequence-length", type=int, default=1024)
+    parser.add_argument("--num-train-epochs", type=float, default=2.0)
+    parser.add_argument("--learning-rate", type=float, default=5e-5)
+    parser.add_argument("--gradient-checkpointing", action="store_true")
+    parser.add_argument("--logging-steps", type=int, default=25)
+    parser.add_argument("--save-steps", type=int, default=200)
+    parser.add_argument("--save-total-limit", type=int, default=2)
+
+    parser.add_argument(
+        "--prepare-only",
+        action="store_true",
+        help="Stop after dataset prep/upload; do not start finetuning",
+    )
+    parser.add_argument(
+        "--skip-upload",
+        action="store_true",
+        help="Prepare local dataset artifacts but skip HF push",
+    )
+    parser.add_argument("--run-id", default=None, help="Optional explicit run id")
+
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    ensure_dirs()
+
+    token = args.hf_token
+    if not token:
+        import os
+
+        token = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN")
+    args.hf_token = token
+
+    run_id = args.run_id or f"SS-FT-{utc_now().strftime('%Y%m%d-%H%M%S')}"
+    run_output_dir = TRAINING_RUNS_DIR / run_id
+    run_output_dir.mkdir(parents=True, exist_ok=True)
+
+    gold_path = Path(args.gold_file).expanduser().resolve()
+    if not gold_path.exists():
+        raise FileNotFoundError(f"Gold file not found: {gold_path}")
+
+    manifest = load_manifest()
+    if not manifest:
+        raise RuntimeError("Manifest is empty. Run 01_register_sources.py and set rights_class first.")
+
+    gold_records = load_gold_records(gold_path)
+    if not gold_records:
+        raise RuntimeError("Gold set is empty or unreadable; cannot fine-tune.")
+
+    examples, stats = build_examples(
+        gold_records=gold_records,
+        manifest=manifest,
+        rights_class_filter=args.rights_class,
+        allow_mixed_rights=args.allow_mixed_rights,
+    )
+
+    if not examples:
+        raise RuntimeError(f"No valid training examples after governance filters. Stats: {stats}")
+
+    rights_for_run = stats.get("rights_seen", [])
+    if not args.allow_mixed_rights and len(rights_for_run) > 1:
+        raise RuntimeError(
+            f"Mixed rights classes in run: {rights_for_run}. This violates governance by default."
+        )
+
+    # Governance metadata
+    gold_hash = sha256_file(gold_path)
+    text_concat = "\n".join(f"{e['book_id']}|{e['page']}|{e['text']}" for e in examples)
+    dataset_hash = sha256_text(text_concat)
+    config_version = "ss_surya_finetune_config_v0.1"
+    if args.pretrained_checkpoint_path:
+        ck_hash = sha256_text(args.pretrained_checkpoint_path)[:8]
+        config_version = f"{config_version}_ckpt_{ck_hash}"
+
+    metadata = {
+        "run_id": run_id,
+        "created_at": utc_iso(),
+        "config_version": config_version,
+        "schema_version": "ss_surya_ocr_finetune_dataset_v1",
+        "gold_file": str(gold_path),
+        "gold_hash": gold_hash,
+        "dataset_hash": dataset_hash,
+        "dataset_size": len(examples),
+        "rights_seen": rights_for_run,
+        "rights_filter": args.rights_class,
+        "dataset_repo_id": args.dataset_repo_id,
+        "model_repo_id": args.model_repo_id,
+        "pretrained_checkpoint_path": args.pretrained_checkpoint_path or "base",
+        "operator": args.operator,
+    }
+
+    dataset_artifact_dir = save_training_artifacts(run_id, examples, stats, metadata)
+    print(f"\nPrepared dataset artifacts: {dataset_artifact_dir}")
+    print(f"Examples kept: {len(examples)} | Rights: {rights_for_run} | Gold hash: {gold_hash[:12]}...")
+
+    summary = {
+        "metadata": metadata,
+        "stats": stats,
+        "train_command": None,
+        "train_returncode": None,
+        "train_stdout_path": None,
+        "train_stderr_path": None,
+    }
+
+    if not args.skip_upload:
+        if not args.hf_token:
+            raise RuntimeError("HF token required for upload. Pass --hf-token or set HF_TOKEN.")
+
+        ds = build_hf_dataset(examples)
+        push_dataset_to_hub(ds, args.dataset_repo_id, args.private_dataset, args.hf_token, run_id)
+        print(f"Pushed dataset to HF: {args.dataset_repo_id}")
+    else:
+        print("Skipped HF upload (--skip-upload).")
+
+    if args.prepare_only:
+        print("Prepare-only mode complete. Finetuning not started.")
+    else:
+        if args.skip_upload:
+            raise RuntimeError(
+                "Cannot start finetuning with --skip-upload because Surya expects --dataset_name. "
+                "Upload dataset first or run with --prepare-only."
+            )
+
+        if not args.hf_token:
+            raise RuntimeError("HF token required for model push during finetuning.")
+
+        train_cmd = build_train_command(args, run_output_dir)
+        stdout_path = run_output_dir / "finetune_stdout.log"
+        stderr_path = run_output_dir / "finetune_stderr.log"
+
+        summary["train_command"] = train_cmd
+        summary["train_stdout_path"] = str(stdout_path)
+        summary["train_stderr_path"] = str(stderr_path)
+
+        print("Launching Surya finetune...")
+        print(" ".join(train_cmd))
+        with open(stdout_path, "w", encoding="utf-8") as out, open(stderr_path, "w", encoding="utf-8") as err:
+            proc = subprocess.run(train_cmd, stdout=out, stderr=err, text=True)
+        summary["train_returncode"] = proc.returncode
+
+        if proc.returncode != 0:
+            raise RuntimeError(
+                f"Surya finetune failed with return code {proc.returncode}. "
+                f"See {stdout_path} and {stderr_path}."
+            )
+
+        print(f"Finetune complete. Model pushed to: {args.model_repo_id}")
+
+    summary_path = TRAINING_RUNS_DIR / f"{run_id}_summary.json"
+    write_run_summary(summary_path, summary)
+
+    notes = {
+        "model_version": args.pretrained_checkpoint_path or "base",
+        "dataset_size": len(examples),
+        "gold_hash": gold_hash,
+        "dataset_repo": args.dataset_repo_id,
+        "model_repo": args.model_repo_id,
+    }
+
+    append_run_log(
+        {
+            "run_id": run_id,
+            "date": utc_now().date().isoformat(),
+            "operator": args.operator,
+            "config_version": config_version,
+            "schema_version": "ss_surya_ocr_finetune_dataset_v1",
+            "source_batch": args.rights_class or ",".join(rights_for_run),
+            "pages_processed": str(len(examples)),
+            "errors": str(
+                stats["skipped_missing_book"]
+                + stats["skipped_missing_manifest"]
+                + stats["skipped_status"]
+                + stats["skipped_blocked_rights"]
+                + stats["skipped_rights_filter"]
+                + stats["skipped_missing_text"]
+                + stats["skipped_missing_image"]
+            ),
+            "cost_usd": "",
+            "output_path": args.model_repo_id,
+            "notes": json.dumps(notes, ensure_ascii=False),
+        }
+    )
+
+    print(f"Run summary: {summary_path}")
+    print(f"Governance log updated: {RUN_LOG_CSV}")
+
+
+if __name__ == "__main__":
+    main()

smoke_signal/scripts/10_recalibrate.py

@@ -0,0 +1,455 @@
+#!/usr/bin/env python3
+"""
+Smoke Signal — Stage 12: Confidence Recalibration
+==================================================
+Recomputes confidence thresholds from the full gold set using a simple
+precision/recall threshold analysis per region class.
+
+Outputs:
+  - manifest/confidence_calibration.json
+  - training/runs/<RUN_ID>_recalibration.json
+  - manifest/run_log.csv entry (governance)
+"""
+
+import argparse
+import csv
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+ROOT = Path(__file__).resolve().parents[1]
+MANIFEST_CSV = ROOT / "manifest" / "source_manifest.csv"
+CALIBRATION_JSON = ROOT / "manifest" / "confidence_calibration.json"
+RUN_LOG_CSV = ROOT / "manifest" / "run_log.csv"
+GOLD_FILE = ROOT / "gold" / "gold_corrections.jsonl"
+RUNS_DIR = ROOT / "training" / "runs"
+
+ELIGIBLE_RIGHTS = {"public-domain", "licensed-owned", "controlled-internal"}
+BLOCKED_RIGHTS = {"unknown", "excluded"}
+
+RUN_LOG_FIELDS = [
+    "run_id",
+    "date",
+    "operator",
+    "config_version",
+    "schema_version",
+    "source_batch",
+    "pages_processed",
+    "errors",
+    "cost_usd",
+    "output_path",
+    "notes",
+]
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def utc_iso() -> str:
+    return utc_now().isoformat().replace("+00:00", "Z")
+
+
+def ensure_run_dirs() -> None:
+    RUNS_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def ensure_run_log() -> None:
+    RUN_LOG_CSV.parent.mkdir(parents=True, exist_ok=True)
+    if RUN_LOG_CSV.exists():
+        return
+    with open(RUN_LOG_CSV, "w", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=RUN_LOG_FIELDS)
+        writer.writeheader()
+
+
+def append_run_log(row: Dict[str, str]) -> None:
+    ensure_run_log()
+    with open(RUN_LOG_CSV, "a", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=RUN_LOG_FIELDS)
+        writer.writerow({k: row.get(k, "") for k in RUN_LOG_FIELDS})
+
+
+def load_manifest() -> Dict[str, Dict[str, str]]:
+    out: Dict[str, Dict[str, str]] = {}
+    if not MANIFEST_CSV.exists():
+        return out
+    with open(MANIFEST_CSV, newline="", encoding="utf-8") as f:
+        for row in csv.DictReader(f):
+            book_id = str(row.get("book_id", "")).strip()
+            if book_id:
+                out[book_id] = row
+    return out
+
+
+def load_gold_records(path: Path) -> List[Dict]:
+    rows: List[Dict] = []
+    if not path.exists():
+        return rows
+    with open(path, encoding="utf-8") as f:
+        for idx, line in enumerate(f, start=1):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                rec = json.loads(line)
+                rec["_line"] = idx
+                rows.append(rec)
+            except json.JSONDecodeError:
+                continue
+    return rows
+
+
+def parse_bool(value) -> Optional[bool]:
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return None
+    text = str(value).strip().lower()
+    if text in {"true", "1", "yes", "y"}:
+        return True
+    if text in {"false", "0", "no", "n"}:
+        return False
+    return None
+
+
+def parse_confidence(value) -> Optional[float]:
+    try:
+        conf = float(value)
+        return max(0.0, min(1.0, conf))
+    except (TypeError, ValueError):
+        return None
+
+
+def infer_was_correct(rec: Dict) -> Optional[bool]:
+    explicit = parse_bool(rec.get("was_correct"))
+    if explicit is not None:
+        return explicit
+
+    final_text = str(rec.get("final_text", "")).strip()
+    raw_text = str(rec.get("raw_text", rec.get("raw_ocr", ""))).strip()
+    if final_text and raw_text:
+        return final_text == raw_text
+    return None
+
+
+def precision_recall_at_threshold(rows: List[Tuple[float, int]], threshold: float) -> Dict[str, float]:
+    tp = fp = fn = tn = 0
+    for conf, label in rows:
+        pred = 1 if conf >= threshold else 0
+        if pred == 1 and label == 1:
+            tp += 1
+        elif pred == 1 and label == 0:
+            fp += 1
+        elif pred == 0 and label == 1:
+            fn += 1
+        else:
+            tn += 1
+
+    precision = tp / (tp + fp) if (tp + fp) else 0.0
+    recall = tp / (tp + fn) if (tp + fn) else 0.0
+    return {
+        "tp": tp,
+        "fp": fp,
+        "fn": fn,
+        "tn": tn,
+        "precision": precision,
+        "recall": recall,
+    }
+
+
+def f_beta(precision: float, recall: float, beta: float) -> float:
+    if precision <= 0 and recall <= 0:
+        return 0.0
+    beta2 = beta * beta
+    denom = (beta2 * precision) + recall
+    if denom <= 0:
+        return 0.0
+    return (1 + beta2) * (precision * recall) / denom
+
+
+def select_thresholds(
+    rows: List[Tuple[float, int]],
+    auto_precision_target: float,
+    auto_recall_floor: float,
+    review_recall_target: float,
+    review_precision_floor: float,
+    quarantine_gap: float,
+) -> Dict:
+    unique_thresholds = sorted({round(conf, 4) for conf, _ in rows})
+    if not unique_thresholds:
+        return {
+            "auto_accept": 0.85,
+            "review": 0.60,
+            "quarantine": 0.35,
+            "metrics": {},
+        }
+
+    # Add boundary values so we can always compute a fallback.
+    thresholds = sorted(set([0.0, 1.0] + unique_thresholds))
+
+    # Auto-accept: highest threshold meeting strict precision target.
+    auto_t = None
+    for t in thresholds:
+        m = precision_recall_at_threshold(rows, t)
+        if m["precision"] >= auto_precision_target and m["recall"] >= auto_recall_floor:
+            auto_t = t
+    if auto_t is None:
+        # Fallback: maximize F0.5 to prioritize precision.
+        auto_t = max(thresholds, key=lambda t: f_beta(
+            precision_recall_at_threshold(rows, t)["precision"],
+            precision_recall_at_threshold(rows, t)["recall"],
+            beta=0.5,
+        ))
+
+    # Review threshold: below/at auto threshold, try to capture most true positives.
+    review_candidates = [t for t in thresholds if t <= auto_t]
+    review_t = None
+    for t in review_candidates:
+        m = precision_recall_at_threshold(rows, t)
+        if m["recall"] >= review_recall_target and m["precision"] >= review_precision_floor:
+            review_t = t
+            break
+    if review_t is None:
+        # Fallback: maximize F1 while respecting t <= auto_t.
+        review_t = max(review_candidates, key=lambda t: f_beta(
+            precision_recall_at_threshold(rows, t)["precision"],
+            precision_recall_at_threshold(rows, t)["recall"],
+            beta=1.0,
+        ))
+
+    review_t = min(review_t, auto_t)
+
+    quarantine_t = max(0.0, review_t - quarantine_gap)
+    quarantine_t = min(quarantine_t, review_t)
+
+    # Round for readability and stable diffs.
+    auto_t = round(float(auto_t), 3)
+    review_t = round(float(review_t), 3)
+    quarantine_t = round(float(quarantine_t), 3)
+
+    # Guarantee monotonic order.
+    if review_t > auto_t:
+        review_t = auto_t
+    if quarantine_t > review_t:
+        quarantine_t = review_t
+
+    return {
+        "auto_accept": auto_t,
+        "review": review_t,
+        "quarantine": quarantine_t,
+        "metrics": {
+            "auto": precision_recall_at_threshold(rows, auto_t),
+            "review": precision_recall_at_threshold(rows, review_t),
+            "quarantine": precision_recall_at_threshold(rows, quarantine_t),
+        },
+    }
+
+
+def build_region_rows(
+    gold_records: List[Dict],
+    manifest: Dict[str, Dict[str, str]],
+    rights_class_filter: Optional[str],
+) -> Tuple[Dict[str, List[Tuple[float, int]]], Dict[str, int]]:
+    by_region: Dict[str, List[Tuple[float, int]]] = {}
+    counters = {
+        "input": len(gold_records),
+        "used": 0,
+        "skipped_missing_book": 0,
+        "skipped_missing_manifest": 0,
+        "skipped_blocked_rights": 0,
+        "skipped_rights_filter": 0,
+        "skipped_missing_confidence": 0,
+        "skipped_missing_label": 0,
+    }
+
+    for rec in gold_records:
+        book_id = str(rec.get("book_id", "")).strip()
+        if not book_id:
+            counters["skipped_missing_book"] += 1
+            continue
+
+        manifest_row = manifest.get(book_id)
+        if not manifest_row:
+            counters["skipped_missing_manifest"] += 1
+            continue
+
+        rights = str(manifest_row.get("rights_class", "unknown")).strip().lower()
+        if rights in BLOCKED_RIGHTS or rights not in ELIGIBLE_RIGHTS:
+            counters["skipped_blocked_rights"] += 1
+            continue
+
+        if rights_class_filter and rights != rights_class_filter:
+            counters["skipped_rights_filter"] += 1
+            continue
+
+        conf = parse_confidence(rec.get("confidence"))
+        if conf is None:
+            counters["skipped_missing_confidence"] += 1
+            continue
+
+        was_correct = infer_was_correct(rec)
+        if was_correct is None:
+            counters["skipped_missing_label"] += 1
+            continue
+
+        # Positive class = OCR output was correct.
+        label = 1 if was_correct else 0
+        region_class = str(rec.get("region_class", "narration")).strip() or "narration"
+
+        by_region.setdefault(region_class, []).append((conf, label))
+        by_region.setdefault("_default", []).append((conf, label))
+        counters["used"] += 1
+
+    return by_region, counters
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Smoke Signal — Stage 12: Recalibration")
+    parser.add_argument("--gold-file", default=str(GOLD_FILE), help="Path to gold corrections JSONL")
+    parser.add_argument(
+        "--rights-class",
+        default=None,
+        choices=sorted(ELIGIBLE_RIGHTS),
+        help="Optional rights class filter",
+    )
+    parser.add_argument("--operator", default="codex", help="Operator for governance log")
+
+    parser.add_argument("--auto-precision-target", type=float, default=0.98)
+    parser.add_argument("--auto-recall-floor", type=float, default=0.20)
+    parser.add_argument("--review-recall-target", type=float, default=0.90)
+    parser.add_argument("--review-precision-floor", type=float, default=0.60)
+    parser.add_argument("--quarantine-gap", type=float, default=0.20)
+    parser.add_argument("--min-samples-per-class", type=int, default=10)
+
+    parser.add_argument("--run-id", default=None, help="Optional explicit run id")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    ensure_run_dirs()
+
+    run_id = args.run_id or f"SS-CAL-{utc_now().strftime('%Y%m%d-%H%M%S')}"
+
+    gold_path = Path(args.gold_file).expanduser().resolve()
+    if not gold_path.exists():
+        raise FileNotFoundError(f"Gold file not found: {gold_path}")
+
+    manifest = load_manifest()
+    if not manifest:
+        raise RuntimeError("Manifest is empty. Cannot enforce rights controls.")
+
+    gold_records = load_gold_records(gold_path)
+    if not gold_records:
+        raise RuntimeError("No valid gold records found.")
+
+    by_region, counters = build_region_rows(gold_records, manifest, args.rights_class)
+    if counters["used"] == 0:
+        raise RuntimeError(f"No usable records for recalibration after filtering. Counters: {counters}")
+
+    calibration: Dict[str, Dict] = {}
+    report_regions: Dict[str, Dict] = {}
+
+    for region_class, rows in by_region.items():
+        if len(rows) < args.min_samples_per_class and region_class != "_default":
+            # Too little data for a reliable per-class threshold; defer to default.
+            continue
+
+        selected = select_thresholds(
+            rows=rows,
+            auto_precision_target=args.auto_precision_target,
+            auto_recall_floor=args.auto_recall_floor,
+            review_recall_target=args.review_recall_target,
+            review_precision_floor=args.review_precision_floor,
+            quarantine_gap=args.quarantine_gap,
+        )
+
+        corrections = sum(1 for _, label in rows if label == 0)
+        calibration[region_class] = {
+            "auto_accept": selected["auto_accept"],
+            "review": selected["review"],
+            "quarantine": selected["quarantine"],
+            "corrections": corrections,
+        }
+
+        report_regions[region_class] = {
+            "samples": len(rows),
+            "correct": sum(1 for _, label in rows if label == 1),
+            "incorrect": sum(1 for _, label in rows if label == 0),
+            "thresholds": calibration[region_class],
+            "metrics": selected["metrics"],
+        }
+
+    # Ensure required defaults exist for runtime readers.
+    if "_default" not in calibration:
+        calibration["_default"] = {
+            "auto_accept": 0.85,
+            "review": 0.60,
+            "quarantine": 0.35,
+            "corrections": 0,
+        }
+
+    default_entry = calibration["_default"]
+    for cls in ["narration", "dialogue-speech-bubble", "caption", "title", "sign-label"]:
+        if cls not in calibration:
+            calibration[cls] = dict(default_entry)
+
+    CALIBRATION_JSON.parent.mkdir(parents=True, exist_ok=True)
+    with open(CALIBRATION_JSON, "w", encoding="utf-8") as f:
+        json.dump(calibration, f, indent=2, ensure_ascii=False)
+
+    report = {
+        "run_id": run_id,
+        "generated_at": utc_iso(),
+        "config_version": "ss_confidence_calibration_v0.1",
+        "schema_version": "ss_confidence_calibration_report_v1",
+        "gold_file": str(gold_path),
+        "rights_class_filter": args.rights_class,
+        "counters": counters,
+        "regions": report_regions,
+        "output_file": str(CALIBRATION_JSON),
+    }
+
+    report_path = RUNS_DIR / f"{run_id}_recalibration.json"
+    with open(report_path, "w", encoding="utf-8") as f:
+        json.dump(report, f, indent=2, ensure_ascii=False)
+
+    append_run_log(
+        {
+            "run_id": run_id,
+            "date": utc_now().date().isoformat(),
+            "operator": args.operator,
+            "config_version": "ss_confidence_calibration_v0.1",
+            "schema_version": "ss_confidence_calibration_report_v1",
+            "source_batch": args.rights_class or "auto",
+            "pages_processed": str(counters["used"]),
+            "errors": str(
+                counters["skipped_missing_book"]
+                + counters["skipped_missing_manifest"]
+                + counters["skipped_blocked_rights"]
+                + counters["skipped_rights_filter"]
+                + counters["skipped_missing_confidence"]
+                + counters["skipped_missing_label"]
+            ),
+            "cost_usd": "",
+            "output_path": str(CALIBRATION_JSON.relative_to(ROOT)),
+            "notes": json.dumps(
+                {
+                    "auto_precision_target": args.auto_precision_target,
+                    "review_recall_target": args.review_recall_target,
+                    "regions_calibrated": sorted(report_regions.keys()),
+                },
+                ensure_ascii=False,
+            ),
+        }
+    )
+
+    print(f"Recalibration complete: {CALIBRATION_JSON}")
+    print(f"Report: {report_path}")
+    print(f"Governance log updated: {RUN_LOG_CSV}")
+
+
+if __name__ == "__main__":
+    main()

smoke_signal_tab.py

@@ -824,7 +824,9 @@ def run_ocr() -> tuple:
                     "book_id": book_id, "filename": row["filename"],
                     "page": page_num, "region_id": region_id,
                     "region_class": "narration",
+                    "rights_class": row.get("rights_class", "unknown"),
                     "crop_path": page_data.get("render_path",""),
+                    "page_image_path": page_data.get("render_path",""),
                     "raw_ocr": raw_text,
                     "confidence": conf, "confidence_class": conf_class,
                     "status": "quarantine" if conf_class == "quarantine" else "pending",
@@ -967,8 +969,10 @@ def save_review_decision(idx: int, final_text: str, action: str, reviewer: str,
         f.write(json.dumps({
             **decision,
             "region_class":  region_class,
+            "rights_class":  item.get("rights_class", "unknown"),
             "confidence":    float(item.get("confidence", 0)),
             "conf_class":    item.get("confidence_class",""),
+            "page_image_path": item.get("page_image_path", item.get("crop_path", "")),
         }) + "\n")
 
     cal         = load_calibration()