Stages 9+10: Codex exporter and validation report

smoke_signal/scripts/07_codex_exporter.py

@@ -0,0 +1,457 @@
+#!/usr/bin/env python3
+"""
+Smoke Signal — Stage 9: Codex Exporter
+========================================
+Takes approved review decisions and exports clean, schema-valid
+JSONL/CSV/Markdown packages ready for Codex ingestion.
+
+Safety rules (non-negotiable):
+  - NEVER export unreviewed low-confidence text
+  - NEVER mix rights classes in the same export batch
+  - NEVER export quarantined or rejected pages
+  - ALWAYS link every exported line back to source PDF hash + page
+  - ALWAYS validate against schema before writing export
+
+Output formats:
+  - codex_export.jsonl  — primary machine-readable format
+  - codex_export.csv    — human-readable review copy
+  - codex_proof.md      — Markdown proof pack for QA
+
+Usage:
+    python scripts/07_codex_exporter.py
+    python scripts/07_codex_exporter.py --book-id SS-BOOK-0001
+    python scripts/07_codex_exporter.py --batch-id SS-BATCH-001
+    python scripts/07_codex_exporter.py --dry-run
+"""
+
+import argparse
+import csv
+import json
+import sys
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+# ── Paths ──────────────────────────────────────────────────────────────────────
+ROOT          = Path(__file__).resolve().parents[1]
+MANIFEST_CSV  = ROOT / "manifest" / "source_manifest.csv"
+REGIONS_DIR   = ROOT / "regions"
+CLEANED_DIR   = ROOT / "regions" / "cleaned"
+REVIEW_DIR    = ROOT / "review"
+EXPORTS_DIR   = ROOT / "exports"
+LOGS_DIR      = ROOT / "logs"
+SCHEMAS_DIR   = ROOT / "schemas"
+
+EXPORTS_DIR.mkdir(parents=True, exist_ok=True)
+LOGS_DIR.mkdir(parents=True, exist_ok=True)
+
+DECISIONS_CSV = REVIEW_DIR / "review_decisions.csv"
+
+# ── Config ────────────────────────────────────────────────────────────────────
+CONFIG = {
+    "config_version":       "ss_export_v0.1",
+    "schema_version":       "1.0",
+    "approved_statuses":    ["accepted", "edited"],
+    "excluded_statuses":    ["rejected", "quarantined", "illustration-only"],
+    "excluded_region_classes": [
+        "page-number", "copyright-legal", "publisher-imprint", "decorative-uncertain"
+    ],
+    "story_region_classes": [
+        "narration", "dialogue-speech-bubble", "caption",
+        "title", "subtitle", "sign-label"
+    ],
+    "eligible_rights":      ["public-domain", "licensed-owned", "controlled-internal"],
+}
+
+EXPORT_FIELDS = [
+    "book_id", "source_id", "source_hash", "page_number", "region_id",
+    "page_class", "region_class", "text_raw", "text_clean", "text_final",
+    "uncertain_words", "confidence", "review_status", "exclusion_reason",
+    "extraction_method", "config_version", "schema_version", "run_id",
+]
+
+
+# ── Manifest I/O ───────────────────────────────────────────────────────────────
+def load_manifest() -> dict:
+    records = {}
+    if not MANIFEST_CSV.exists():
+        return records
+    with open(MANIFEST_CSV, newline="", encoding="utf-8") as f:
+        for row in csv.DictReader(f):
+            if row.get("book_id"):
+                records[row["book_id"]] = row
+    return records
+
+
+def save_manifest(records: dict) -> None:
+    fields = [
+        "book_id", "source_id", "filename", "sha256", "file_size_bytes",
+        "page_count", "rights_class", "source_location", "acquisition_date",
+        "status", "allowed_use", "notes"
+    ]
+    rows = sorted(records.values(), key=lambda r: r.get("book_id", ""))
+    with open(MANIFEST_CSV, "w", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=fields)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+# ── Load review decisions ─────────────────────────────────────────────────────
+def load_decisions() -> dict:
+    decisions = {}
+    if not DECISIONS_CSV.exists():
+        return decisions
+    with open(DECISIONS_CSV, newline="", encoding="utf-8") as f:
+        for row in csv.DictReader(f):
+            decisions[row.get("region_id", "")] = row
+    return decisions
+
+
+# ── Load region data ───────────────────────────────────────────────────────────
+def load_regions(book_id: str) -> Optional[dict]:
+    """Load cleaned regions if available, otherwise ordered regions."""
+    cleaned = CLEANED_DIR / f"{book_id}_cleaned_regions.json"
+    if cleaned.exists():
+        return json.load(open(cleaned))
+    ordered = REGIONS_DIR / f"{book_id}_ordered_regions.json"
+    if ordered.exists():
+        return json.load(open(ordered))
+    return None
+
+
+# ── Schema validator ──────────────────────────────────────────────────────────
+def validate_record(record: dict) -> list:
+    """Basic validation — returns list of errors."""
+    errors = []
+    required = ["book_id", "source_hash", "page_number", "region_id",
+                 "region_class", "text_final", "review_status", "run_id"]
+    for field in required:
+        if not record.get(field):
+            errors.append(f"missing_required_field: {field}")
+
+    if record.get("confidence") is not None:
+        try:
+            conf = float(record["confidence"])
+            if not (0.0 <= conf <= 1.0):
+                errors.append(f"confidence_out_of_range: {conf}")
+        except (ValueError, TypeError):
+            errors.append("confidence_not_numeric")
+
+    valid_statuses = CONFIG["approved_statuses"] + ["excluded"]
+    if record.get("review_status") not in valid_statuses:
+        errors.append(f"invalid_review_status: {record.get('review_status')}")
+
+    return errors
+
+
+# ── Build export record ────────────────────────────────────────────────────────
+def build_export_record(
+    region: dict,
+    page_data: dict,
+    manifest_record: dict,
+    decision: Optional[dict],
+    run_id: str,
+) -> Optional[dict]:
+    """
+    Build a single Codex export record from region + decision data.
+    Returns None if the record should be excluded.
+    """
+    region_class = region.get("region_class", "decorative-uncertain")
+    region_id    = region.get("region_id", "")
+    page_num     = region.get("page_number") or page_data.get("page_number")
+
+    # Exclude non-story regions
+    if region_class in CONFIG["excluded_region_classes"]:
+        return None
+
+    # Check decision status
+    if decision:
+        review_status = decision.get("status", "pending")
+        if review_status not in CONFIG["approved_statuses"]:
+            return None
+        text_final = decision.get("final_text") or region.get("text_clean") or region.get("text", "")
+    else:
+        # No decision yet — exclude from export
+        return None
+
+    # Determine extraction method
+    if region.get("llm_model"):
+        method = "ocr+llm"
+    elif region.get("text_clean") and region.get("text_clean") != region.get("text"):
+        method = "ocr+llm"
+    else:
+        method = "ocr"
+
+    if page_data.get("route") == "embedded_text":
+        method = "embedded-text"
+
+    record = {
+        "book_id":          manifest_record.get("book_id", ""),
+        "source_id":        manifest_record.get("source_id", ""),
+        "source_hash":      manifest_record.get("sha256", ""),
+        "page_number":      page_num,
+        "region_id":        region_id,
+        "page_class":       page_data.get("page_class", "story-page"),
+        "region_class":     region_class,
+        "text_raw":         region.get("text", ""),
+        "text_clean":       region.get("text_clean") or region.get("text", ""),
+        "text_final":       text_final.strip(),
+        "uncertain_words":  json.dumps(region.get("uncertain_words", [])),
+        "confidence":       round(float(region.get("confidence", 0)), 4),
+        "review_status":    review_status,
+        "exclusion_reason": "",
+        "extraction_method": method,
+        "config_version":   CONFIG["config_version"],
+        "schema_version":   CONFIG["schema_version"],
+        "run_id":           run_id,
+    }
+
+    return record
+
+
+# ── Per-book export ───────────────────────────────────────────────────────────
+def export_book(
+    manifest_record: dict,
+    decisions: dict,
+    run_id: str,
+    dry_run: bool,
+) -> tuple:
+    book_id     = manifest_record["book_id"]
+    filename    = manifest_record["filename"]
+    rights      = manifest_record.get("rights_class", "unknown")
+    source_hash = manifest_record.get("sha256", "")
+
+    print(f"\n  [{book_id}] {filename}")
+
+    # Rights check
+    if rights not in CONFIG["eligible_rights"]:
+        print(f"    ✗ Skipped — rights_class={rights} not eligible for export")
+        return [], {"error": "rights_not_eligible"}
+
+    # Load region data
+    regions_data = load_regions(book_id)
+    if regions_data is None:
+        print(f"    ✗ No region data found. Run earlier stages first.")
+        return [], {"error": "no_region_data"}
+
+    export_records = []
+    skipped        = 0
+    validation_errors = []
+
+    for page_data in regions_data.get("pages", []):
+        page_num = page_data.get("page_number")
+        route    = page_data.get("route", "ocr")
+
+        for region in page_data.get("regions", []):
+            region_id = region.get("region_id", "")
+            decision  = decisions.get(region_id)
+
+            record = build_export_record(
+                region, page_data, manifest_record, decision, run_id
+            )
+
+            if record is None:
+                skipped += 1
+                continue
+
+            # Validate
+            errors = validate_record(record)
+            if errors:
+                validation_errors.append({
+                    "region_id": region_id,
+                    "errors": errors
+                })
+                print(f"    ⚠️  Validation failed for {region_id}: {errors}")
+                continue
+
+            export_records.append(record)
+
+    print(f"    Exported: {len(export_records)} records | Skipped: {skipped} | Validation errors: {len(validation_errors)}")
+
+    if validation_errors:
+        print(f"    ⚠️  {len(validation_errors)} records failed validation — excluded from export")
+
+    return export_records, {}
+
+
+# ── Write export files ────────────────────────────────────────────────────────
+def write_exports(
+    all_records: list,
+    run_id: str,
+    batch_id: str,
+    dry_run: bool,
+) -> dict:
+    if not all_records:
+        print("\n  No records to export.")
+        return {}
+
+    ts = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+    batch_tag = batch_id or ts
+
+    # ── JSONL ────────────────────────────────────────────────────────────────
+    jsonl_path = EXPORTS_DIR / f"codex_export_{batch_tag}.jsonl"
+    if not dry_run:
+        with open(jsonl_path, "w", encoding="utf-8") as f:
+            for rec in all_records:
+                f.write(json.dumps(rec, ensure_ascii=False) + "\n")
+        print(f"\n  JSONL → {jsonl_path.relative_to(ROOT)}")
+
+    # ── CSV ──────────────────────────────────────────────────────────────────
+    csv_path = EXPORTS_DIR / f"codex_export_{batch_tag}.csv"
+    if not dry_run:
+        with open(csv_path, "w", newline="", encoding="utf-8") as f:
+            writer = csv.DictWriter(f, fieldnames=EXPORT_FIELDS)
+            writer.writeheader()
+            writer.writerows(all_records)
+        print(f"  CSV   → {csv_path.relative_to(ROOT)}")
+
+    # ── Markdown proof pack ──────────────────────────────────────────────────
+    md_path = EXPORTS_DIR / f"codex_proof_{batch_tag}.md"
+    if not dry_run:
+        # Group by book
+        by_book = {}
+        for rec in all_records:
+            bid = rec["book_id"]
+            by_book.setdefault(bid, []).append(rec)
+
+        lines = [
+            f"# Smoke Signal — Codex Export Proof Pack",
+            f"**Batch:** {batch_tag}  ",
+            f"**Run ID:** {run_id}  ",
+            f"**Exported:** {datetime.utcnow().isoformat()}Z  ",
+            f"**Records:** {len(all_records)}  ",
+            f"**Config:** {CONFIG['config_version']}  ",
+            f"\n---\n",
+        ]
+
+        for bid, records in sorted(by_book.items()):
+            lines.append(f"## {bid}")
+            # Group by page
+            by_page = {}
+            for rec in records:
+                p = rec["page_number"]
+                by_page.setdefault(p, []).append(rec)
+
+            for page_num in sorted(by_page.keys()):
+                lines.append(f"\n### Page {page_num}")
+                for rec in by_page[page_num]:
+                    status_icon = "✓" if rec["review_status"] == "accepted" else "✎"
+                    lines.append(f"\n**[{status_icon} {rec['region_class']}]** `conf:{rec['confidence']:.0%}`")
+                    lines.append(f"\n> {rec['text_final']}\n")
+                    if rec.get("uncertain_words") and rec["uncertain_words"] != "[]":
+                        lines.append(f"*Uncertain words: {rec['uncertain_words']}*\n")
+            lines.append("\n---\n")
+
+        with open(md_path, "w", encoding="utf-8") as f:
+            f.write("\n".join(lines))
+        print(f"  MD    → {md_path.relative_to(ROOT)}")
+
+    # ── Export manifest ──────────────────────────────────────────────────────
+    manifest_path = EXPORTS_DIR / f"export_manifest_{batch_tag}.json"
+    export_manifest = {
+        "run_id":        run_id,
+        "batch_id":      batch_tag,
+        "exported_at":   datetime.utcnow().isoformat() + "Z",
+        "config_version": CONFIG["config_version"],
+        "schema_version": CONFIG["schema_version"],
+        "record_count":  len(all_records),
+        "books":         list({r["book_id"] for r in all_records}),
+        "files": {
+            "jsonl": str(jsonl_path.relative_to(ROOT)) if not dry_run else None,
+            "csv":   str(csv_path.relative_to(ROOT)) if not dry_run else None,
+            "proof": str(md_path.relative_to(ROOT)) if not dry_run else None,
+        }
+    }
+
+    if not dry_run:
+        with open(manifest_path, "w") as f:
+            json.dump(export_manifest, f, indent=2)
+        print(f"  Manifest → {manifest_path.relative_to(ROOT)}")
+
+    return export_manifest
+
+
+# ── Main ───────────────────────────────────────────────────────────────────────
+def main():
+    parser = argparse.ArgumentParser(description="Smoke Signal — Stage 9: Codex Exporter")
+    parser.add_argument("--book-id",  help="Export a single book by ID")
+    parser.add_argument("--batch-id", help="Tag this export batch (e.g. SS-BATCH-001)")
+    parser.add_argument("--dry-run",  action="store_true")
+    args = parser.parse_args()
+
+    run_id  = f"SS-RUN-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}"
+    dry_run = args.dry_run
+
+    print(f"\n{'='*60}")
+    print(f"  Smoke Signal — Stage 9: Codex Exporter")
+    print(f"  Run ID    : {run_id}")
+    print(f"  Config    : {CONFIG['config_version']}")
+    print(f"  Schema    : {CONFIG['schema_version']}")
+    if dry_run:
+        print(f"  Mode      : DRY RUN")
+    print(f"{'='*60}")
+
+    manifest  = load_manifest()
+    decisions = load_decisions()
+
+    if not manifest:
+        print("\n  Manifest empty. Run earlier stages first.")
+        sys.exit(1)
+
+    if not decisions:
+        print("\n  No review decisions found.")
+        print("  Run the review workbench (06_review_workbench.py) first.")
+        sys.exit(1)
+
+    print(f"\n  Review decisions loaded: {len(decisions)}")
+    approved = sum(1 for d in decisions.values() if d.get("status") in CONFIG["approved_statuses"])
+    print(f"  Approved decisions: {approved}")
+
+    # Select books
+    if args.book_id:
+        books = [manifest[args.book_id]] if args.book_id in manifest else []
+    else:
+        # Export books that have region data
+        books = []
+        for bid, rec in manifest.items():
+            if load_regions(bid) is not None:
+                if rec.get("rights_class") in CONFIG["eligible_rights"]:
+                    books.append(rec)
+
+    if not books:
+        print("\n  No eligible books found for export.")
+        sys.exit(0)
+
+    print(f"  Books to export: {len(books)}")
+
+    all_records = []
+    t_start     = time.time()
+
+    for record in books:
+        records, error = export_book(record, decisions, run_id, dry_run)
+        all_records.extend(records)
+
+    # Write all exports
+    export_manifest = write_exports(all_records, run_id, args.batch_id, dry_run)
+
+    # Update manifest status to exported
+    if not dry_run and all_records:
+        exported_books = {r["book_id"] for r in all_records}
+        for bid in exported_books:
+            if bid in manifest:
+                manifest[bid]["status"] = "exported"
+        save_manifest(manifest)
+
+    elapsed = round(time.time() - t_start, 1)
+
+    print(f"\n{'─'*60}")
+    print(f"  Total records exported : {len(all_records)}")
+    print(f"  Time                   : {elapsed}s")
+    print(f"{'─'*60}")
+    print(f"\n  ✓ Export complete.")
+    print(f"  Next: Run validation report (Stage 10) before Codex ingestion.\n")
+
+
+if __name__ == "__main__":
+    main()

smoke_signal/scripts/08_validation_report.py

@@ -0,0 +1,412 @@
+#!/usr/bin/env python3
+"""
+Smoke Signal — Stage 10: Validation Report
+============================================
+Generates a full pipeline validation report before Codex ingestion.
+
+Checks:
+  - Source manifest completeness and hash integrity
+  - Page routing accuracy (spot check)
+  - OCR confidence distribution
+  - Review queue clearance rate
+  - Export schema validity
+  - Contamination risk (low-confidence text in export)
+  - Rights class separation
+  - Page traceability (every exported line has source evidence)
+
+Outputs:
+  - reports/validation_report_<batch>.md  — human-readable
+  - reports/validation_report_<batch>.json — machine-readable
+
+Usage:
+    python scripts/08_validation_report.py
+    python scripts/08_validation_report.py --batch-id SS-BATCH-001
+    python scripts/08_validation_report.py --export-file exports/codex_export_20260516.jsonl
+"""
+
+import argparse
+import csv
+import json
+import sys
+from datetime import datetime
+from pathlib import Path
+
+# ── Paths ──────────────────────────────────────────────────────────────────────
+ROOT          = Path(__file__).resolve().parents[1]
+MANIFEST_CSV  = ROOT / "manifest" / "source_manifest.csv"
+EXPORTS_DIR   = ROOT / "exports"
+REGIONS_DIR   = ROOT / "regions"
+OCR_RAW_DIR   = ROOT / "ocr_raw"
+REVIEW_DIR    = ROOT / "review"
+REPORTS_DIR   = ROOT / "reports"
+LOGS_DIR      = ROOT / "logs"
+
+REPORTS_DIR.mkdir(parents=True, exist_ok=True)
+DECISIONS_CSV = REVIEW_DIR / "review_decisions.csv"
+QUEUE_CSV     = REVIEW_DIR / "review_queue.csv"
+
+CONFIG = {
+    "config_version":              "ss_validate_v0.1",
+    "min_page_route_accuracy":     0.95,
+    "min_ocr_word_accuracy":       0.80,
+    "min_review_clearance":        0.90,
+    "max_contamination_risk":      0.00,
+    "confidence_contamination_threshold": 0.60,
+}
+
+
+# ── Loaders ────────────────────────────────────────────────────────────────────
+def load_manifest() -> list:
+    if not MANIFEST_CSV.exists():
+        return []
+    with open(MANIFEST_CSV, newline="", encoding="utf-8") as f:
+        return list(csv.DictReader(f))
+
+
+def load_queue() -> list:
+    if not QUEUE_CSV.exists():
+        return []
+    with open(QUEUE_CSV, newline="", encoding="utf-8") as f:
+        return list(csv.DictReader(f))
+
+
+def load_decisions() -> list:
+    if not DECISIONS_CSV.exists():
+        return []
+    with open(DECISIONS_CSV, newline="", encoding="utf-8") as f:
+        return list(csv.DictReader(f))
+
+
+def load_export(export_file: Path) -> list:
+    records = []
+    if not export_file.exists():
+        return records
+    with open(export_file, encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if line:
+                try:
+                    records.append(json.loads(line))
+                except json.JSONDecodeError:
+                    pass
+    return records
+
+
+# ── Check functions ────────────────────────────────────────────────────────────
+def check_manifest(manifest: list) -> dict:
+    total       = len(manifest)
+    unknown_rights = sum(1 for r in manifest if r.get("rights_class") == "unknown")
+    excluded    = sum(1 for r in manifest if r.get("rights_class") == "excluded")
+    no_hash     = sum(1 for r in manifest if not r.get("sha256"))
+    status_counts = {}
+    for r in manifest:
+        s = r.get("status", "unknown")
+        status_counts[s] = status_counts.get(s, 0) + 1
+
+    issues = []
+    if unknown_rights > 0:
+        issues.append(f"{unknown_rights} sources have unknown rights class")
+    if no_hash > 0:
+        issues.append(f"{no_hash} sources missing SHA-256 hash")
+
+    return {
+        "check": "source_manifest",
+        "total_sources":    total,
+        "unknown_rights":   unknown_rights,
+        "excluded":         excluded,
+        "no_hash":          no_hash,
+        "status_counts":    status_counts,
+        "issues":           issues,
+        "passed":           len(issues) == 0,
+    }
+
+
+def check_review_clearance(queue: list, decisions: list) -> dict:
+    total_queued  = len(queue)
+    decided_ids   = {d.get("region_id") for d in decisions}
+    queue_ids     = {q.get("region_id") for q in queue}
+    cleared       = len(queue_ids & decided_ids)
+    pending       = total_queued - cleared
+    clearance_rate = cleared / max(total_queued, 1)
+
+    approved  = sum(1 for d in decisions if d.get("status") in ("accepted", "edited"))
+    rejected  = sum(1 for d in decisions if d.get("status") == "rejected")
+    quarantined = sum(1 for d in decisions if d.get("status") == "quarantined")
+
+    passed = clearance_rate >= CONFIG["min_review_clearance"]
+    issues = []
+    if pending > 0:
+        issues.append(f"{pending} items still pending review ({clearance_rate:.0%} cleared)")
+
+    return {
+        "check":           "review_clearance",
+        "total_queued":    total_queued,
+        "cleared":         cleared,
+        "pending":         pending,
+        "clearance_rate":  round(clearance_rate, 3),
+        "approved":        approved,
+        "rejected":        rejected,
+        "quarantined":     quarantined,
+        "target":          CONFIG["min_review_clearance"],
+        "issues":          issues,
+        "passed":          passed,
+    }
+
+
+def check_export_contamination(export_records: list) -> dict:
+    """Check for low-confidence text that shouldn't be in export."""
+    threshold   = CONFIG["confidence_contamination_threshold"]
+    risky       = []
+
+    for rec in export_records:
+        conf = float(rec.get("confidence", 1.0))
+        status = rec.get("review_status", "")
+        if conf < threshold and status == "accepted":
+            risky.append({
+                "region_id": rec.get("region_id"),
+                "confidence": conf,
+                "review_status": status,
+            })
+
+    contamination_risk = len(risky) / max(len(export_records), 1)
+    passed = contamination_risk <= CONFIG["max_contamination_risk"]
+
+    issues = []
+    if risky:
+        issues.append(f"{len(risky)} low-confidence records in export (conf < {threshold})")
+
+    return {
+        "check":               "contamination_risk",
+        "total_export_records": len(export_records),
+        "risky_records":       len(risky),
+        "contamination_rate":  round(contamination_risk, 4),
+        "threshold":           threshold,
+        "risky_sample":        risky[:5],
+        "issues":              issues,
+        "passed":              passed,
+    }
+
+
+def check_traceability(export_records: list) -> dict:
+    """Every exported record must have source_hash, page_number, region_id, run_id."""
+    missing_trace = []
+    for rec in export_records:
+        missing = [f for f in ("source_hash", "page_number", "region_id", "run_id")
+                   if not rec.get(f)]
+        if missing:
+            missing_trace.append({
+                "region_id": rec.get("region_id", "unknown"),
+                "missing_fields": missing
+            })
+
+    passed = len(missing_trace) == 0
+    issues = []
+    if missing_trace:
+        issues.append(f"{len(missing_trace)} records missing traceability fields")
+
+    return {
+        "check":                "traceability",
+        "total_records":        len(export_records),
+        "untraceable_records":  len(missing_trace),
+        "sample":               missing_trace[:5],
+        "issues":               issues,
+        "passed":               passed,
+    }
+
+
+def check_rights_separation(export_records: list, manifest: list) -> dict:
+    """Confirm rights classes are not mixed in export."""
+    manifest_by_id = {r.get("book_id"): r for r in manifest}
+    rights_in_export = set()
+    unknown_in_export = []
+
+    for rec in export_records:
+        bid = rec.get("book_id")
+        m   = manifest_by_id.get(bid, {})
+        rc  = m.get("rights_class", "unknown")
+        rights_in_export.add(rc)
+        if rc in ("unknown", "excluded"):
+            unknown_in_export.append(bid)
+
+    issues = []
+    if unknown_in_export:
+        issues.append(f"Unknown/excluded rights class found in export: {set(unknown_in_export)}")
+
+    return {
+        "check":                "rights_separation",
+        "rights_classes_found": list(rights_in_export),
+        "unknown_in_export":    list(set(unknown_in_export)),
+        "issues":               issues,
+        "passed":               len(unknown_in_export) == 0,
+    }
+
+
+def check_ocr_confidence_distribution(queue: list) -> dict:
+    """Summarise OCR confidence distribution across reviewed pages."""
+    confs = []
+    for item in queue:
+        try:
+            confs.append(float(item.get("confidence", 0)))
+        except (ValueError, TypeError):
+            pass
+
+    if not confs:
+        return {"check": "ocr_confidence", "passed": True, "note": "no_data"}
+
+    confs.sort()
+    n       = len(confs)
+    mean    = sum(confs) / n
+    below60 = sum(1 for c in confs if c < 0.60)
+    below80 = sum(1 for c in confs if c < 0.80)
+
+    return {
+        "check":       "ocr_confidence_distribution",
+        "page_count":  n,
+        "mean":        round(mean, 3),
+        "min":         round(confs[0], 3),
+        "max":         round(confs[-1], 3),
+        "p25":         round(confs[n // 4], 3),
+        "p50":         round(confs[n // 2], 3),
+        "p75":         round(confs[3 * n // 4], 3),
+        "below_0.60":  below60,
+        "below_0.80":  below80,
+        "passed":      True,
+    }
+
+
+# ── Report writer ───────────────────────────────────────────────────────��─────
+def write_report(checks: list, batch_tag: str, export_path: Path) -> Path:
+    all_passed = all(c.get("passed", False) for c in checks)
+    issues_all = [i for c in checks for i in c.get("issues", [])]
+    verdict    = "✅ PASS — Safe to proceed to Codex ingestion" if all_passed else "❌ FAIL — Do not ingest until issues resolved"
+
+    # ── JSON ─────────────────────────────────────────────────────────────────
+    json_path = REPORTS_DIR / f"validation_report_{batch_tag}.json"
+    report_data = {
+        "batch_id":      batch_tag,
+        "generated_at":  datetime.utcnow().isoformat() + "Z",
+        "config_version": CONFIG["config_version"],
+        "export_file":   str(export_path),
+        "verdict":       "PASS" if all_passed else "FAIL",
+        "all_issues":    issues_all,
+        "checks":        checks,
+    }
+    with open(json_path, "w") as f:
+        json.dump(report_data, f, indent=2)
+
+    # ── Markdown ──────────────────────────────────────────────────────────────
+    md_path = REPORTS_DIR / f"validation_report_{batch_tag}.md"
+    lines   = [
+        f"# Smoke Signal — Validation Report",
+        f"**Batch:** {batch_tag}  ",
+        f"**Generated:** {datetime.utcnow().isoformat()}Z  ",
+        f"**Export:** `{export_path.name}`  ",
+        f"\n## Verdict\n",
+        f"### {verdict}\n",
+    ]
+
+    if issues_all:
+        lines.append("## Issues to Resolve\n")
+        for issue in issues_all:
+            lines.append(f"- ⚠️ {issue}")
+        lines.append("")
+
+    lines.append("## Check Results\n")
+    lines.append("| Check | Status | Notes |")
+    lines.append("|-------|--------|-------|")
+
+    for c in checks:
+        icon  = "✅" if c.get("passed") else "❌"
+        name  = c.get("check", "unknown").replace("_", " ").title()
+        notes = "; ".join(c.get("issues", [])) or "OK"
+        lines.append(f"| {name} | {icon} | {notes} |")
+
+    lines.append("\n## Detailed Results\n")
+    for c in checks:
+        lines.append(f"### {c.get('check','').replace('_',' ').title()}")
+        for k, v in c.items():
+            if k not in ("check", "issues", "passed", "sample", "risky_sample"):
+                lines.append(f"- **{k}:** {v}")
+        lines.append("")
+
+    lines.append("---")
+    lines.append(f"*Smoke Signal {CONFIG['config_version']} · Generated {datetime.utcnow().date()}*")
+
+    with open(md_path, "w", encoding="utf-8") as f:
+        f.write("\n".join(lines))
+
+    return md_path, json_path, all_passed
+
+
+# ── Main ───────────────────────────────────────────────────────────────────────
+def main():
+    parser = argparse.ArgumentParser(description="Smoke Signal — Stage 10: Validation Report")
+    parser.add_argument("--batch-id",    help="Batch ID tag")
+    parser.add_argument("--export-file", help="Path to JSONL export file to validate")
+    args = parser.parse_args()
+
+    batch_tag = args.batch_id or datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+
+    print(f"\n{'='*60}")
+    print(f"  Smoke Signal — Stage 10: Validation Report")
+    print(f"  Batch   : {batch_tag}")
+    print(f"  Config  : {CONFIG['config_version']}")
+    print(f"{'='*60}\n")
+
+    # Find export file
+    if args.export_file:
+        export_path = Path(args.export_file)
+    else:
+        jsonl_files = sorted(EXPORTS_DIR.glob("codex_export_*.jsonl"), key=lambda p: p.stat().st_mtime, reverse=True)
+        if not jsonl_files:
+            print("  No export files found. Run 07_codex_exporter.py first.")
+            sys.exit(1)
+        export_path = jsonl_files[0]
+        print(f"  Using latest export: {export_path.name}\n")
+
+    # Load data
+    manifest       = load_manifest()
+    queue          = load_queue()
+    decisions      = load_decisions()
+    export_records = load_export(export_path)
+
+    print(f"  Sources in manifest : {len(manifest)}")
+    print(f"  Review queue items  : {len(queue)}")
+    print(f"  Review decisions    : {len(decisions)}")
+    print(f"  Export records      : {len(export_records)}\n")
+
+    # Run checks
+    checks = [
+        check_manifest(manifest),
+        check_review_clearance(queue, decisions),
+        check_ocr_confidence_distribution(queue),
+        check_export_contamination(export_records),
+        check_traceability(export_records),
+        check_rights_separation(export_records, manifest),
+    ]
+
+    for c in checks:
+        icon = "✅" if c.get("passed") else "❌"
+        print(f"  {icon} {c['check']}")
+        for issue in c.get("issues", []):
+            print(f"     ⚠️  {issue}")
+
+    # Write reports
+    md_path, json_path, all_passed = write_report(checks, batch_tag, export_path)
+
+    print(f"\n{'─'*60}")
+    print(f"  Report MD   → {md_path.relative_to(ROOT)}")
+    print(f"  Report JSON → {json_path.relative_to(ROOT)}")
+    print(f"{'─'*60}")
+
+    if all_passed:
+        print(f"\n  ✅ ALL CHECKS PASSED — Approval Gate E cleared.")
+        print(f"     Safe to ingest into Codex.\n")
+    else:
+        print(f"\n  ❌ CHECKS FAILED — Do not ingest until issues are resolved.")
+        print(f"     Fix the issues above and re-run this report.\n")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()