code/self_healing_finetuning/data_pipeline/learning_pdf_to_jsonl.py

from __future__ import annotations

import hashlib
import json
import re
from pathlib import Path
from typing import Any

from n21.config import load_structured, write_json
from n21.settings import CONFIG_ROOT, REPO_ROOT
from observability.audit_log import utc_now
from data_pipeline.pdf_warning_filter import suppress_known_pypdf_pointer_noise
from data_pipeline.repair_coverage import DEFAULT_MINIMUMS, classify_repair_row


SYSTEM_TEMPLATE = (
    "You are Linvest21_FinGPT acting as an institutional {asset_class} {role}. "
    "Learn from curated source material, preserve evidence discipline, separate fact from inference, "
    "and avoid investment advice."
)


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def clean_text(text: str) -> str:
    text = text.replace("\x00", " ")
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


def load_role_manifest(role_dir: Path) -> dict[str, dict[str, Any]]:
    manifest_path = role_dir / "manifest.json"
    if not manifest_path.exists():
        return {}
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    items = manifest.get("items", [])
    return {str(item.get("filename")): item for item in items if item.get("filename")}


def extract_pages(pdf_path: Path) -> list[dict[str, Any]]:
    try:
        from pypdf import PdfReader
    except ModuleNotFoundError as exc:
        raise RuntimeError("Missing dependency pypdf. Install with: python -m pip install pypdf") from exc

    with suppress_known_pypdf_pointer_noise():
        reader = PdfReader(str(pdf_path))
        pages: list[dict[str, Any]] = []
        for index, page in enumerate(reader.pages, start=1):
            try:
                text = clean_text(page.extract_text() or "")
            except Exception as exc:  # pypdf can fail on malformed individual pages.
                text = f"[page extraction failed: {type(exc).__name__}: {exc}]"
            if text and not text.startswith("[page extraction failed:"):
                pages.append({"page": index, "text": text})
    return pages


def chunk_pages(pages: list[dict[str, Any]], *, chunk_chars: int) -> list[dict[str, Any]]:
    chunks: list[dict[str, Any]] = []
    current: list[str] = []
    page_start: int | None = None
    page_end: int | None = None

    def flush() -> None:
        nonlocal current, page_start, page_end
        if not current or page_start is None or page_end is None:
            return
        chunks.append({"page_start": page_start, "page_end": page_end, "text": clean_text("\n\n".join(current))})
        current = []
        page_start = None
        page_end = None

    for page in pages:
        page_no = int(page["page"])
        text = str(page["text"])
        if page_start is None:
            page_start = page_no
        if sum(len(part) for part in current) + len(text) > chunk_chars and current:
            flush()
            page_start = page_no
        current.append(f"[Page {page_no}]\n{text}")
        page_end = page_no
    flush()
    return chunks


def build_record(
    *,
    asset_class: str,
    role: str,
    pdf_path: Path,
    pdf_sha256: str,
    item: dict[str, Any] | None,
    chunk: dict[str, Any],
    chunk_index: int,
    chunk_count: int,
) -> dict[str, Any]:
    title = str((item or {}).get("title") or pdf_path.stem.replace("_", " "))
    author = str((item or {}).get("author") or "unknown")
    license_status = str((item or {}).get("license") or "unknown")
    excerpt = str(chunk["text"])
    user_prompt = (
        "Convert the following curated equity research source excerpt into reusable Linvest21 research knowledge. "
        "Focus on analytical principles, valuation/research process, evidence use, and risk framing.\n\n"
        f"Source title: {title}\n"
        f"Author/source: {author}\n"
        f"Pages: {chunk['page_start']}-{chunk['page_end']}\n\n"
        f"Excerpt:\n{excerpt}"
    )
    assistant_response = (
        f"Source: {title}\n"
        f"Role lens: Linvest21 {asset_class} {role}\n"
        f"Evidence pages: {chunk['page_start']}-{chunk['page_end']}\n\n"
        "Reusable research knowledge:\n"
        f"{excerpt}\n\n"
        "Application guardrails:\n"
        "- Use this material as research context, not as standalone investment advice.\n"
        "- Preserve source attribution when using the idea in a research memo.\n"
        "- Separate observed facts, author claims, and Linvest21 inference."
    )
    return {
        "messages": [
            {"role": "system", "content": SYSTEM_TEMPLATE.format(asset_class=asset_class, role=role.replace("_", " "))},
            {"role": "user", "content": user_prompt},
            {"role": "assistant", "content": assistant_response},
        ],
        "metadata": {
            "asset_class": asset_class,
            "role": role,
            "task": "curated_financial_research_sft",
            "source_pdf": pdf_path.name,
            "source_sha256": pdf_sha256,
            "source_title": title,
            "source_author": author,
            "license": license_status,
            "page_start": chunk["page_start"],
            "page_end": chunk["page_end"],
            "chunk_index": chunk_index,
            "chunk_count": chunk_count,
            "curated_root": "data/learning",
            "policy": "curated_learning_material",
            "created_at": utc_now(),
        },
    }


def chunk_text(text: str, *, chunk_chars: int) -> list[dict[str, Any]]:
    chunks: list[dict[str, Any]] = []
    clean = clean_text(text)
    if not clean:
        return []
    for index, start in enumerate(range(0, len(clean), chunk_chars), start=1):
        chunk = clean[start:start + chunk_chars]
        chunks.append({"section_start": index, "section_end": index, "text": clean_text(chunk)})
    return chunks


def build_normalized_record(
    *,
    asset_class: str,
    role: str,
    normalized_path: Path,
    normalized_sha256: str,
    normalized: dict[str, Any],
    chunk: dict[str, Any],
    chunk_index: int,
    chunk_count: int,
) -> dict[str, Any]:
    title = str(normalized.get("source_title") or normalized_path.stem.replace("_", " "))
    source_format = str(normalized.get("format") or "normalized")
    license_status = str(normalized.get("license_status") or "unknown")
    excerpt = str(chunk["text"])
    user_prompt = (
        "Convert the following normalized Linvest21 source excerpt into reusable investment research knowledge. "
        "Focus on analytical principles, evidence use, risk framing, and role-specific workflow.\n\n"
        f"Source title: {title}\n"
        f"Source format: {source_format}\n"
        f"Section: {chunk['section_start']}-{chunk['section_end']}\n\n"
        f"Excerpt:\n{excerpt}"
    )
    assistant_response = (
        f"Source: {title}\n"
        f"Role lens: Linvest21 {asset_class} {role}\n"
        f"Evidence section: {chunk['section_start']}-{chunk['section_end']}\n\n"
        "Reusable research knowledge:\n"
        f"{excerpt}\n\n"
        "Application guardrails:\n"
        "- Use this material as research context, not as standalone investment advice.\n"
        "- Preserve source attribution when using the idea in a research memo.\n"
        "- Separate observed facts, author claims, and Linvest21 inference."
    )
    return {
        "messages": [
            {"role": "system", "content": SYSTEM_TEMPLATE.format(asset_class=asset_class, role=role.replace("_", " "))},
            {"role": "user", "content": user_prompt},
            {"role": "assistant", "content": assistant_response},
        ],
        "metadata": {
            "asset_class": asset_class,
            "role": role,
            "task": "normalized_financial_research_sft",
            "source_normalized": normalized_path.name,
            "source_sha256": normalized_sha256,
            "source_title": title,
            "source_url": normalized.get("source_url"),
            "source_format": source_format,
            "license": license_status,
            "section_start": chunk["section_start"],
            "section_end": chunk["section_end"],
            "chunk_index": chunk_index,
            "chunk_count": chunk_count,
            "curated_root": "data/learning",
            "policy": "normalized_policy_approved_material",
            "created_at": utc_now(),
        },
    }


def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(json.dumps(row, ensure_ascii=True, sort_keys=True) for row in rows) + "\n", encoding="utf-8")


def convert_learning_role(
    *,
    asset_class: str,
    role: str,
    chunk_chars: int | None = None,
    min_text_chars: int | None = None,
    skip_existing: bool = True,
) -> dict[str, Any]:
    config = load_structured(CONFIG_ROOT / "data" / "learning_corpus.json")
    curated_root = REPO_ROOT / str(config["curated_root"])
    if asset_class not in config["asset_classes"]:
        raise ValueError(f"unsupported asset_class {asset_class}; expected one of {config['asset_classes']}")
    if role not in config["roles"]:
        raise ValueError(f"unsupported role {role}; expected one of {config['roles']}")

    chunk_chars = int(chunk_chars or config["jsonl_generation"]["default_chunk_chars"])
    min_text_chars = int(min_text_chars or config["jsonl_generation"]["default_min_text_chars"])
    role_dir = curated_root / asset_class / role
    if not role_dir.exists():
        raise FileNotFoundError(f"role directory not found: {role_dir}")

    manifest_items = load_role_manifest(role_dir)
    pdfs = sorted(role_dir.glob("*.pdf"))
    normalized_sources = sorted(role_dir.glob("*.normalized.json"))
    combined_rows: list[dict[str, Any]] = []
    pdf_reports: list[dict[str, Any]] = []
    normalized_reports: list[dict[str, Any]] = []

    for pdf_path in pdfs:
        pdf_sha = sha256_file(pdf_path)
        output_path = pdf_path.with_name(f"{pdf_path.stem}{config['jsonl_generation']['output_suffix']}")
        invalid_existing_error: str | None = None
        if skip_existing and output_path.exists() and output_path.stat().st_size > 0:
            try:
                rows = load_jsonl(output_path)
            except ValueError as exc:
                invalid_existing_error = str(exc)
            else:
                combined_rows.extend(rows)
                pdf_reports.append(
                    {
                        "pdf": pdf_path.name,
                        "status": "reused_existing_jsonl",
                        "text_chars": None,
                        "page_count_with_text": None,
                        "record_count": len(rows),
                        "output_jsonl": str(output_path),
                        "sha256": pdf_sha,
                    }
                )
                continue

        pages = extract_pages(pdf_path)
        text_chars = sum(len(page["text"]) for page in pages)
        if text_chars < min_text_chars:
            if output_path.exists():
                output_path.unlink()
            pdf_reports.append(
                {
                    "pdf": pdf_path.name,
                    "status": "skipped_insufficient_extractable_text",
                    "text_chars": text_chars,
                    "page_count_with_text": len(pages),
                    "output_jsonl": None,
                    "sha256": pdf_sha,
                }
            )
            continue
        chunks = chunk_pages(pages, chunk_chars=chunk_chars)
        item = manifest_items.get(pdf_path.name)
        rows = [
            build_record(
                asset_class=asset_class,
                role=role,
                pdf_path=pdf_path,
                pdf_sha256=pdf_sha,
                item=item,
                chunk=chunk,
                chunk_index=index,
                chunk_count=len(chunks),
            )
            for index, chunk in enumerate(chunks, start=1)
        ]
        write_jsonl(output_path, rows)
        combined_rows.extend(rows)
        status = "regenerated_invalid_existing_jsonl" if invalid_existing_error else "converted"
        report_item = {
            "pdf": pdf_path.name,
            "status": status,
            "text_chars": text_chars,
            "page_count_with_text": len(pages),
            "record_count": len(rows),
            "output_jsonl": str(output_path),
            "sha256": pdf_sha,
        }
        if invalid_existing_error:
            report_item["invalid_existing_error"] = invalid_existing_error
        pdf_reports.append(
            report_item
        )

    for normalized_path in normalized_sources:
        normalized_sha = sha256_file(normalized_path)
        output_path = normalized_path.with_name(f"{normalized_path.stem}.hf_finetune.jsonl")
        invalid_existing_error: str | None = None
        if skip_existing and output_path.exists() and output_path.stat().st_size > 0:
            try:
                rows = load_jsonl(output_path)
            except ValueError as exc:
                invalid_existing_error = str(exc)
            else:
                combined_rows.extend(rows)
                normalized_reports.append(
                    {
                        "normalized_source": normalized_path.name,
                        "status": "reused_existing_jsonl",
                        "text_chars": None,
                        "record_count": len(rows),
                        "output_jsonl": str(output_path),
                        "sha256": normalized_sha,
                    }
                )
                continue
        normalized = json.loads(normalized_path.read_text(encoding="utf-8"))
        if not normalized.get("eligibility", {}).get("training"):
            normalized_reports.append(
                {
                    "normalized_source": normalized_path.name,
                    "status": "skipped_not_training_eligible",
                    "text_chars": len(str(normalized.get("text") or "")),
                    "record_count": 0,
                    "output_jsonl": None,
                    "sha256": normalized_sha,
                    "eligibility": normalized.get("eligibility", {}),
                }
            )
            continue
        text = str(normalized.get("text") or "")
        text_chars = len(text)
        if text_chars < min_text_chars:
            normalized_reports.append(
                {
                    "normalized_source": normalized_path.name,
                    "status": "skipped_insufficient_extractable_text",
                    "text_chars": text_chars,
                    "record_count": 0,
                    "output_jsonl": None,
                    "sha256": normalized_sha,
                }
            )
            continue
        chunks = chunk_text(text, chunk_chars=chunk_chars)
        rows = [
            build_normalized_record(
                asset_class=asset_class,
                role=role,
                normalized_path=normalized_path,
                normalized_sha256=normalized_sha,
                normalized=normalized,
                chunk=chunk,
                chunk_index=index,
                chunk_count=len(chunks),
            )
            for index, chunk in enumerate(chunks, start=1)
        ]
        write_jsonl(output_path, rows)
        combined_rows.extend(rows)
        status = "regenerated_invalid_existing_jsonl" if invalid_existing_error else "converted"
        item = {
            "normalized_source": normalized_path.name,
            "status": status,
            "text_chars": text_chars,
            "record_count": len(rows),
            "output_jsonl": str(output_path),
            "sha256": normalized_sha,
        }
        if invalid_existing_error:
            item["invalid_existing_error"] = invalid_existing_error
        normalized_reports.append(item)

    combined_path = role_dir / f"{asset_class}_{role}.hf_finetune.jsonl"
    write_jsonl(combined_path, combined_rows)
    report = {
        "asset_class": asset_class,
        "role": role,
        "role_dir": str(role_dir),
        "combined_jsonl": str(combined_path),
        "pdf_count": len(pdfs),
        "normalized_source_count": len(normalized_sources),
        "converted_pdf_count": sum(1 for item in pdf_reports if item["status"] in {"converted", "regenerated_invalid_existing_jsonl"}),
        "converted_normalized_count": sum(1 for item in normalized_reports if item["status"] in {"converted", "regenerated_invalid_existing_jsonl"}),
        "reused_pdf_count": sum(1 for item in pdf_reports if item["status"] == "reused_existing_jsonl"),
        "reused_normalized_count": sum(1 for item in normalized_reports if item["status"] == "reused_existing_jsonl"),
        "skipped_pdf_count": sum(
            1
            for item in pdf_reports
            if item["status"] not in {"converted", "regenerated_invalid_existing_jsonl", "reused_existing_jsonl"}
        ),
        "skipped_normalized_count": sum(
            1
            for item in normalized_reports
            if item["status"] not in {"converted", "regenerated_invalid_existing_jsonl", "reused_existing_jsonl"}
        ),
        "record_count": len(combined_rows),
        "chunk_chars": chunk_chars,
        "min_text_chars": min_text_chars,
        "schema": "huggingface_chat_messages_jsonl",
        "reports": pdf_reports,
        "normalized_reports": normalized_reports,
        "created_at": utc_now(),
    }
    write_json(role_dir / f"{asset_class}_{role}_conversion_manifest.json", report)
    return report


def load_jsonl(path: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8", newline="") as handle:
        for line_no, line in enumerate(handle, start=1):
            if not line.strip():
                continue
            try:
                item = json.loads(line)
            except json.JSONDecodeError as exc:
                raise ValueError(
                    f"invalid JSONL at {path}:{line_no}: {exc.msg} "
                    f"(column {exc.colno}, char {exc.pos})"
                ) from exc
            if not isinstance(item, dict) or not isinstance(item.get("messages"), list):
                raise ValueError(f"invalid Hugging Face chat JSONL at {path}:{line_no}")
            rows.append(item)
    return rows


def convert_learning_tree(
    *,
    asset_class: str | None = None,
    role: str | None = None,
    source_type: str = "pdf",
    chunk_chars: int | None = None,
    min_text_chars: int | None = None,
    skip_existing: bool = True,
) -> dict[str, Any]:
    if source_type.lower() != "pdf":
        raise ValueError(f"unsupported source_type for now: {source_type}; currently supported: pdf")
    config = load_structured(CONFIG_ROOT / "data" / "learning_corpus.json")
    asset_classes = [asset_class] if asset_class else list(config["asset_classes"])
    roles = [role] if role else list(config["roles"])
    reports: list[dict[str, Any]] = []
    for asset in asset_classes:
        for role_name in roles:
            role_dir = REPO_ROOT / str(config["curated_root"]) / asset / role_name
            if not role_dir.exists():
                continue
            reports.append(
                convert_learning_role(
                    asset_class=asset,
                    role=role_name,
                    chunk_chars=chunk_chars,
                    min_text_chars=min_text_chars,
                    skip_existing=skip_existing,
                )
            )
    return {
        "source_type": source_type,
        "asset_class": asset_class or "all",
        "role": role or "all",
        "role_count": len(reports),
        "pdf_count": sum(int(item["pdf_count"]) for item in reports),
        "normalized_source_count": sum(int(item.get("normalized_source_count", 0)) for item in reports),
        "converted_pdf_count": sum(int(item["converted_pdf_count"]) for item in reports),
        "converted_normalized_count": sum(int(item.get("converted_normalized_count", 0)) for item in reports),
        "reused_pdf_count": sum(int(item.get("reused_pdf_count", 0)) for item in reports),
        "reused_normalized_count": sum(int(item.get("reused_normalized_count", 0)) for item in reports),
        "skipped_pdf_count": sum(int(item["skipped_pdf_count"]) for item in reports),
        "skipped_normalized_count": sum(int(item.get("skipped_normalized_count", 0)) for item in reports),
        "record_count": sum(int(item["record_count"]) for item in reports),
        "reports": reports,
        "created_at": utc_now(),
    }


def build_training_jsonl_from_learning(
    *,
    output_path: Path,
    source: Path | None = None,
    asset_class: str | None = None,
    role: str | None = None,
    repair_oversample_factor: int = 1,
    max_repair_selected_ratio: float | None = None,
) -> dict[str, Any]:
    config = load_structured(CONFIG_ROOT / "data" / "learning_corpus.json")
    curated_root = REPO_ROOT / str(config["curated_root"])
    if source is None:
        source_path = curated_root.resolve()
    elif source.is_absolute():
        source_path = source.resolve()
    else:
        source_path = (REPO_ROOT / source).resolve()
    output_path = output_path.resolve()
    suffix = str(config["jsonl_generation"]["output_suffix"])

    if source_path.is_file():
        jsonls = [source_path]
    elif source_path.is_dir():
        jsonls = sorted(source_path.rglob(f"*{suffix}"))
        combined_names = {f"{asset}_{role_name}{suffix}" for asset in config["asset_classes"] for role_name in config["roles"]}
        jsonls = [
            item
            for item in jsonls
            if item.name not in combined_names
        ]
    else:
        raise FileNotFoundError(f"learning source path not found: {source_path}")

    selected: list[Path] = []
    rows: list[dict[str, Any]] = []
    for jsonl in jsonls:
        loaded = load_jsonl(jsonl)
        if asset_class or role:
            filtered = [
                row
                for row in loaded
                if (asset_class is None or row.get("metadata", {}).get("asset_class") == asset_class)
                and (role is None or row.get("metadata", {}).get("role") == role)
            ]
        else:
            filtered = loaded
        if not filtered:
            continue
        selected.append(jsonl)
        rows.extend(filtered)

    if not rows:
        raise ValueError(f"no training rows selected from {source_path}")

    repair_oversample_factor = max(1, int(repair_oversample_factor or 1))
    source_record_count = len(rows)
    repair_classifications = [classify_repair_row(row) for row in rows]
    source_repair_row_count = sum(1 for classes in repair_classifications if classes)
    repair_cap_applied = max_repair_selected_ratio is not None
    selected_repair_source_rows = source_repair_row_count
    dropped_repair_source_rows = 0
    if max_repair_selected_ratio is not None:
        max_repair_selected_ratio = float(max_repair_selected_ratio)
        if max_repair_selected_ratio <= 0 or max_repair_selected_ratio >= 1:
            raise ValueError("--max-repair-selected-ratio must be greater than 0 and less than 1")
        nonrepair_rows: list[dict[str, Any]] = []
        repair_candidates: list[tuple[int, dict[str, Any], set[str]]] = []
        for index, (row, classes) in enumerate(zip(rows, repair_classifications)):
            if classes:
                repair_candidates.append((index, row, classes))
            else:
                nonrepair_rows.append(row)

        selected_indexes: set[int] = set()
        selected_counts = {key: 0 for key in DEFAULT_MINIMUMS}
        for repair_class, minimum in DEFAULT_MINIMUMS.items():
            for index, _row, classes in repair_candidates:
                if selected_counts[repair_class] * repair_oversample_factor >= minimum:
                    break
                if repair_class not in classes:
                    continue
                selected_indexes.add(index)
                for item in classes:
                    if item in selected_counts:
                        selected_counts[item] += 1

        missing = {
            key: {
                "selected_base_rows": value,
                "effective_rows_after_oversample": value * repair_oversample_factor,
                "minimum": DEFAULT_MINIMUMS[key],
            }
            for key, value in selected_counts.items()
            if value * repair_oversample_factor < DEFAULT_MINIMUMS[key]
        }
        if missing:
            raise ValueError(
                "repair cap cannot be applied because selected repair rows cannot satisfy coverage minimums: "
                + json.dumps(missing, sort_keys=True)
            )

        selected_repair_rows = [row for index, row, _classes in repair_candidates if index in selected_indexes]
        effective_repair_rows = len(selected_repair_rows) * repair_oversample_factor
        final_record_count = len(nonrepair_rows) + effective_repair_rows
        repair_ratio = 0 if final_record_count == 0 else effective_repair_rows / final_record_count
        if repair_ratio > max_repair_selected_ratio:
            raise ValueError(
                "repair cap cannot be applied while preserving repair coverage: "
                f"effective_repair_rows={effective_repair_rows}, nonrepair_rows={len(nonrepair_rows)}, "
                f"ratio={repair_ratio:.6f}, max_repair_selected_ratio={max_repair_selected_ratio:.6f}. "
                "Add more original non-repair training rows or lower repair coverage minimums only by explicit policy."
            )

        rows = nonrepair_rows + selected_repair_rows
        repair_classifications = [classify_repair_row(row) for row in rows]
        selected_repair_source_rows = len(selected_repair_rows)
        dropped_repair_source_rows = source_repair_row_count - selected_repair_source_rows

    original_record_count = len(rows)
    repair_row_count = 0
    oversampled_rows: list[dict[str, Any]] = []
    if repair_oversample_factor > 1:
        for row in rows:
            repair_classes = sorted(classify_repair_row(row))
            if not repair_classes:
                continue
            repair_row_count += 1
            for copy_index in range(2, repair_oversample_factor + 1):
                cloned = json.loads(json.dumps(row))
                metadata = cloned.setdefault("metadata", {})
                if isinstance(metadata, dict):
                    metadata["repair_oversample_copy"] = copy_index
                    metadata["repair_oversample_factor"] = repair_oversample_factor
                    metadata["repair_oversample_classes"] = repair_classes
                oversampled_rows.append(cloned)
        rows.extend(oversampled_rows)
    else:
        repair_row_count = sum(1 for row in rows if classify_repair_row(row))

    required_reasoning_files: list[Path] = []
    if asset_class and role:
        role_dir = curated_root / asset_class / role
        if role_dir.exists():
            required_reasoning_files = sorted(role_dir.glob(f"synthetic_{asset_class}_{role}_critical_reasoning{suffix}"))
    selected_resolved = {path.resolve() for path in selected}
    missing_required_reasoning = [path for path in required_reasoning_files if path.resolve() not in selected_resolved]
    if missing_required_reasoning:
        missing = ", ".join(str(path) for path in missing_required_reasoning)
        raise ValueError(f"required role-grounded reasoning JSONL was not selected for training: {missing}")

    write_jsonl(output_path, rows)
    selected_training_sha256 = sha256_file(output_path)
    manifest = {
        "schema_version": "selected_training_manifest_v1",
        "output_jsonl": str(output_path),
        "source": str(source_path),
        "asset_class": asset_class or "all",
        "role": role or "all",
        "input_jsonl_count": len(selected),
        "record_count": len(rows),
        "source_record_count": source_record_count,
        "original_record_count": original_record_count,
        "repair_row_count": repair_row_count,
        "source_repair_row_count": source_repair_row_count,
        "selected_repair_source_rows": selected_repair_source_rows,
        "dropped_repair_source_rows": dropped_repair_source_rows,
        "repair_cap_applied": repair_cap_applied,
        "max_repair_selected_ratio": max_repair_selected_ratio,
        "repair_oversample_factor": repair_oversample_factor,
        "repair_oversampled_record_count": len(oversampled_rows),
        "input_jsonls": [str(path) for path in selected],
        "selected_training_sha256": selected_training_sha256,
        "required_reasoning_jsonls": [str(path) for path in required_reasoning_files],
        "required_reasoning_included": not missing_required_reasoning,
        "created_at": utc_now(),
    }
    write_json(output_path.with_suffix(".manifest.json"), manifest)
    return manifest