ingest/modal_app.py

"""Modal app — heavy ingestion at scale (the Modal-bonus track).

Ingests 3D-printing knowledge from documentation URLs, GitHub repos, slicer
profiles, and HF datasets into the three lanes the Chief Engineer consumes.

THE CONTRACT — three artifact shapes (one JSON object per line):
  A) reference fact     -> data/references.jsonl
     {"material","param","value","source"}                       (hard params)
  B) candidate lesson   -> data/_modal_candidate_lessons.jsonl    (REVIEW before ledger)
     {"job_id","material","geometry_type","env_temp","env_humidity",
      "outcome","lesson","source":"ingested","timestamp"}
  C) calibration obs    -> sim/calibration/observations.modal.jsonl
     {"material","geometry_type","env_temp","env_humidity","nozzle_temp",
      "bed_temp","retraction_mm","fan_pct","first_layer_fan_pct","outcome","quality"?}

Run (after `uv pip install modal datasets` + `modal token set`):
    modal run ingest/modal_app.py                          # all sources
    modal run ingest/modal_app.py --source klipper-config  # single source
    modal run ingest/modal_app.py --category firmware      # category of sources
"""

from __future__ import annotations

import json
import re
import hashlib
from datetime import datetime, timezone
from pathlib import Path

# Import-guarded so the rest of the app/tests never depend on modal being present.
try:
    import modal
except Exception:  # pragma: no cover
    modal = None  # type: ignore

# ── Enums (must match core/models.py) ──────────────────────────────────────────
MATERIALS = ["PLA", "PETG", "ABS", "TPU"]
GEOMETRY_TYPES = ["overhang", "bridge", "stringing", "adhesion", "vase"]
OUTCOMES = ["success", "failed_sag", "failed_stringing"]

# ── Source Registry ───────────────────────────────────────────────────────────
# Each source: key -> (url, category, source_type, description)
# source_type determines how we fetch & parse it:
#   "github_repo"  — clone, scan for config files (*.cfg, *.h, *.ini)
#   "web_doc"      — fetch HTML, extract text, parse for params & lessons
#   "prusa_profiles" — download INI files, parse with Prusa parser
#   "product_spec" — extract material parameters from product pages
#   "research_repo" — clone, look for datasets/configs

SOURCES: dict[str, tuple[str, str, str, str]] = {
    # ── Firmware, Libraries & Host Controllers ──
    "moonraker": (
        "https://github.com/Arksine/moonraker",
        "firmware", "github_repo",
        "Moonraker web API server for Klipper"
    ),
    "mainsail": (
        "https://github.com/meteyou/mainsail",
        "firmware", "github_repo",
        "Mainsail web dashboard for Klipper"
    ),
    "tmcstepper": (
        "https://github.com/teemuatlut/TMCStepper",
        "firmware", "github_repo",
        "TMC stepper driver library"
    ),
    "klipper-thermistors": (
        "https://www.klipper3d.org/Config_Reference.html#common-thermistors",
        "firmware", "web_doc",
        "Klipper thermistor sensor designations"
    ),
    "tmc26x": (
        "https://github.com/trinamic/TMC26XStepper",
        "firmware", "github_repo",
        "TMC26X high-current driver library"
    ),
    "arduino-l6470": (
        "https://github.com/ameyer/Arduino-L6470",
        "firmware", "github_repo",
        "L6470 stepper driver library"
    ),
    "u8glib": (
        "https://github.com/olikraus/U8glib_Arduino",
        "firmware", "github_repo",
        "Display rendering library"
    ),
    "platformio": (
        "http://docs.platformio.org/en/latest/projectconf.html",
        "firmware", "web_doc",
        "PlatformIO project configuration"
    ),

    # ── Calibration Frameworks & Parametric Rules ──
    "reprap-calibration": (
        "http://reprap.org/wiki/Calibration",
        "calibration", "web_doc",
        "RepRap calibration framework"
    ),
    "pid-tuning": (
        "http://reprap.org/wiki/PID_Tuning",
        "calibration", "web_doc",
        "PID tuning for heater performance"
    ),
    "linear-advance": (
        "http://marlinfw.org/docs/features/lin_advance.html",
        "calibration", "web_doc",
        "Marlin Linear Advance configuration"
    ),
    "laser-spindle": (
        "http://marlinfw.org/docs/configuration/laser_spindle.html",
        "calibration", "web_doc",
        "PWM-to-RPM spindle configuration"
    ),
    "probes": (
        "http://marlinfw.org/docs/configuration/probes.html",
        "calibration", "web_doc",
        "Z-probe configuration"
    ),
    "gcode-actions": (
        "https://reprap.org/wiki/G-code#Action_commands",
        "calibration", "web_doc",
        "G-code protocol specification"
    ),
    "jerk-motion": (
        "https://github.com/synthetos/TinyG/wiki/Jerk-Controlled-Motion-Explained",
        "calibration", "web_doc",
        "Jerk-controlled motion kinematics"
    ),
    "junction-deviation": (
        "https://reprap.org/forum/read.php?1,739819",
        "calibration", "web_doc",
        "Junction Deviation equations"
    ),

    # ── Research Datasets & ML Resources ──
    "bcn3d-moveo": (
        "https://www.bcn3d.com/bcn3d-moveo-the-future-of-learning-robotic-arm/",
        "research", "web_doc",
        "BCN3D Moveo robotic arm (3D-ADAM base)"
    ),
    "3dtime-dataloader": (
        "https://github.com/3DTimeDataset/3DTime_pytorch_dataloader",
        "research", "research_repo",
        "3DTime time-series slicing data loader"
    ),
    "klipper-analysis": (
        "https://github.com/worksasintended/klipper_linear_movement_analysis",
        "research", "research_repo",
        "Klipper linear movement analysis"
    ),

    # ── Hardware Profiles & Slicer Configuration ──
    "prusa-anycubic": (
        "https://files.prusa3d.com/wp-content/uploads/repository/PrusaSlicer-settings-master/live/Anycubic/",
        "profiles", "prusa_profiles",
        "PrusaSlicer Anycubic machine profiles"
    ),
    "hartrusion-config": (
        "https://hartrusion.com/en/prusaslicer-config-for-anycubic-4max-pro-2-0/",
        "profiles", "web_doc",
        "Anycubic 4Max Pro 2.0 PrusaSlicer config"
    ),
    "sainsmart-tpu": (
        "https://www.sainsmart.com/collections/tpu-filament/products/all-colors-tpu-flexible-filament-1-75mm-0-8kg-1-76lb",
        "profiles", "product_spec",
        "Sainsmart TPU filament specifications"
    ),
    "ratos-install": (
        "https://os.ratrig.com/docs/installation",
        "profiles", "web_doc",
        "RatOS installation framework"
    ),
    "ratos-upgrade": (
        "https://os.ratrig.com/docs/upgrading_rc3",
        "profiles", "web_doc",
        "RatOS upgrade migration"
    ),

    # ── Structured Profile & Config Repos (Tier S) ──
    "bambu-filament-profiles": (
        "https://github.com/bambulab/BambuStudio",
        "structured", "bambu_json_profiles",
        "BambuStudio tree-structured filament JSON profiles (200+ materials)"
    ),
    "kanrog-klipper-configs": (
        "https://github.com/Kanrog/klipper-config-generator",
        "structured", "klipper_config_repo",
        "Klipper config generator: 150+ motherboard pin maps, PID, max temps"
    ),
    "3dprint-saviour-thresholds": (
        "https://github.com/Manicben/3DPrintSaviour",
        "structured", "failure_detection_repo",
        "3DPrintSaviour: NRMSE failure detection thresholds + classification logic"
    ),
    "jklewa-filament-profiles": (
        "https://github.com/jklewa/filament-profiles-data",
        "structured", "filament_profiles_repo",
        "Community-verified filament profiles: nozzle/bed temps, vendor, price"
    ),
    "fdm-error-detection": (
        "https://github.com/NilsHagenBeyer/3D-printing_recorder",
        "structured", "fdm_error_gcode",
        "FDM error detection: G-code + YAML with known failure outcomes (Lane C goldmine)"
    ),
}

# ── HF Dataset Mappers (for structured datasets) ──────────────────────────────
# These are separate from the documentation sources above.
# Each mapper takes one dataset row and returns zero or more ("A"|"B"|"C", record) tuples.

def _map_3d_adam(row: dict) -> list[tuple[str, dict]]:
    """3D-ADAM defect dataset → Lane C calibration observations.

    The 3D-ADAM dataset (pmchard/3D-ADAM) contains images and defect masks
    for 3D printing defects. We extract defect type → outcome mapping and
    any available print settings to produce calibration observations.

    Dataset structure (from anomalib loader):
      - image: PIL Image of the printed part
      - mask: defect mask
      - label: defect class (warping, under_extrusion, stringing, cracking)
      - category: part category
    """
    records = []
    defect = str(row.get("label", row.get("category", ""))).lower()

    # Map 3D-ADAM defect classes to Chief Engineer outcomes
    defect_to_outcome = {
        "warping": "failed_sag",
        "under_extrusion": "failed_sag",
        "stringing": "failed_stringing",
        "cracking": "failed_sag",
    }
    outcome = defect_to_outcome.get(defect)
    if not outcome:
        return records

    # Map defect to geometry type
    defect_to_geometry = {
        "warping": "adhesion",
        "under_extrusion": "overhang",
        "stringing": "stringing",
        "cracking": "adhesion",
    }
    geometry_type = defect_to_geometry.get(defect, "overhang")

    # 3D-ADAM doesn't include print settings in the dataset itself,
    # but we can emit Lane B lessons from the defect taxonomy.
    # Lane C requires actual settings — only emit if settings columns exist.
    has_settings = all(k in row for k in ("nozzle_temp", "bed_temp", "retraction_mm", "fan_pct"))
    if has_settings:
        records.append(("C", {
            "material": row.get("material", "PLA"),
            "geometry_type": geometry_type,
            "env_temp": float(row.get("env_temp", 22)),
            "env_humidity": float(row.get("env_humidity", 45)),
            "nozzle_temp": float(row["nozzle_temp"]),
            "bed_temp": float(row["bed_temp"]),
            "retraction_mm": float(row["retraction_mm"]),
            "fan_pct": float(row["fan_pct"]),
            "first_layer_fan_pct": float(row.get("first_layer_fan_pct", 0)),
            "outcome": outcome,
            "quality": float(row.get("quality", 0.5)),
        }))

    # Always emit Lane B lesson from defect taxonomy
    defect_lessons = {
        "warping": "Corners lift when lower layers cool and contract — raise bed temp, enclose, slow first layer.",
        "under_extrusion": "Gaps and weak walls from too-low temp or too-fast flow — raise temp or slow down, check for clogs.",
        "stringing": "Fine whiskers across gaps from wet filament or hot travel — dry filament, lower temp, tune retraction.",
        "cracking": "Layers split under stress from over-cooling — reduce fan, raise temp, enclose for ABS.",
    }
    lesson_text = defect_lessons.get(defect, "")
    if lesson_text:
        job_id = f"modal-3dadam-{hashlib.md5(str(row).encode()).hexdigest()[:8]}"
        records.append(("B", {
            "job_id": job_id,
            "material": row.get("material", "PLA"),
            "geometry_type": geometry_type,
            "env_temp": float(row.get("env_temp", 22)),
            "env_humidity": float(row.get("env_humidity", 45)),
            "outcome": outcome,
            "lesson": lesson_text,
            "source": "ingested",
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }))

    return records


def _map_gcode(row: dict) -> list[tuple[str, dict]]:
    """Slicer g-code corpus → Lane A material baselines.

    The ablam/gcode dataset contains G-code files from Printables.
    We parse M104 (set extruder temp), M140 (set bed temp), and
    retraction settings to extract reference facts.
    """
    records = []
    gcode_text = str(row.get("gcode", row.get("content", row.get("text", ""))))
    if not gcode_text:
        return records

    # Extract temperatures from G-code
    nozzle_match = re.search(r"M104\s+S(\d+)", gcode_text)
    bed_match = re.search(r"M140\s+S(\d+)", gcode_text)
    retract_match = re.search(r"G1\s+E-?(\d+\.?\d*).*retract", gcode_text, re.I)

    # Try to determine material from filename or comments
    filename = str(row.get("filename", row.get("file_name", ""))).upper()
    material = "PLA"  # default
    for m in MATERIALS:
        if m in filename:
            material = m
            break

    source = f"ablam/gcode:{row.get('filename', row.get('id', 'unknown'))}"

    if nozzle_match:
        records.append(("A", {
            "material": material,
            "param": "nozzle_temp",
            "value": float(nozzle_match.group(1)),
            "source": source,
        }))
    if bed_match:
        records.append(("A", {
            "material": material,
            "param": "bed_temp",
            "value": float(bed_match.group(1)),
            "source": source,
        }))
    if retract_match:
        records.append(("A", {
            "material": material,
            "param": "retraction_mm",
            "value": float(retract_match.group(1)),
            "source": source,
        }))

    return records


def _map_3dtime(row: dict) -> list[tuple[str, dict]]:
    """3DTime metadata CSV → Lane A reference facts (material, infill, geometry)."""
    records = []
    material_map = {"pla": "PLA", "pet": "PETG", "abs": "ABS", "tpu": "TPU"}
    material = material_map.get(str(row.get("Material", "")).lower(), "PLA")
    source = f"3DTime:{row.get('3D mesh name', row.get('G-code file name', 'unknown'))}"

    for dim, param in [("Bounding box X (mm)", "bbox_x_mm"),
                        ("Bounding box Y (mm)", "bbox_y_mm"),
                        ("Bounding box Z (mm)", "bbox_z_mm")]:
        val = row.get(dim)
        if val:
            try:
                records.append(("A", {"material": material, "param": param,
                                      "value": float(val), "source": source}))
            except ValueError:
                pass

    for key, param in [("Infill density (%)", "infill_density_pct"),
                        ("Infill rotation (°)", "infill_rotation")]:
        val = row.get(key)
        if val:
            try:
                records.append(("A", {"material": material, "param": param,
                                      "value": float(val), "source": source}))
            except ValueError:
                pass

    infill_type = row.get("Infill type", "")
    if infill_type:
        records.append(("A", {"material": material, "param": "infill_type",
                              "value": 0, "source": f"{source}:{infill_type}"}))

    print_time = row.get("Print time (s)")
    if print_time:
        try:
            records.append(("A", {"material": material, "param": "print_time_s",
                                  "value": float(print_time), "source": source}))
        except ValueError:
            pass

    return records


# dataset key → (HF dataset id, mapper)
HF_MAPPERS = {
    "3d-adam": ("pmchard/3D-ADAM", _map_3d_adam),
    "gcode": ("ablam/gcode", _map_gcode),
    "3dtime": ("3DTimeDataset/3DTime", _map_3dtime),
}

_VALID_LANES = {"A", "B", "C"}

# ── Deterministic parsers for documentation content ───────────────────────────

def _material_of(text: str) -> str | None:
    """Detect material name in text."""
    up = text.upper()
    for m in MATERIALS:
        if m in up:
            return m
    return None


def _extract_temperature_values(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Extract temperature reference facts from documentation text.

    Looks for patterns like:
      - "nozzle temperature: 200-220°C"
      - "bed temperature 60°C"
      - "max_temp: 300"
      - "recommended temperature 210°C for PLA"
    """
    records = []
    # Pattern: material name near a temperature value
    # e.g. "PLA at 200°C", "PETG: 230-240°C"
    temp_patterns = [
        # nozzle_temp patterns
        (r'(?:nozzle|hotend|extruder)\s*(?:temp(?:erature)?)?\s*[:=]?\s*(\d{3})(?:\s*[-–]\s*\d{3})?\s*°?[CF]?', "nozzle_temp"),
        (r'(?:print(?:ing)?\s*)?temp(?:erature)?\s*[:=]?\s*(\d{3})(?:\s*[-–]\s*\d{3})?\s*°?[CF]?', "nozzle_temp"),
        # bed_temp patterns
        (r'(?:bed|heatbed)\s*(?:temp(?:erature)?)?\s*[:=]?\s*(\d{2,3})\s*°?[CF]?', "bed_temp"),
        # retraction patterns
        (r'retract(?:ion)?\s*(?:length|distance)?\s*[:=]?\s*(\d+\.?\d*)\s*mm', "retraction_mm"),
        # max_temp patterns
        (r'max(?:imum)?\s*_?temp(?:erature)?\s*[:=]?\s*(\d{3})\s*°?[CF]?', "max_temp"),
        # fan patterns
        (r'(?:part\s*)?(?:cooling\s*)?fan\s*(?:speed|pct|percent)?\s*[:=]?\s*(\d{1,3})\s*%?', "fan_pct"),
    ]

    for pattern, param in temp_patterns:
        for match in re.finditer(pattern, text, re.IGNORECASE):
            value = float(match.group(1))
            # Find nearby material mention (within 200 chars before)
            context_start = max(0, match.start() - 200)
            context = text[context_start:match.end()]
            material = _material_of(context) or "*"
            records.append(("A", {
                "material": material,
                "param": param,
                "value": value,
                "source": source_label,
            }))

    return records


def _extract_lessons_from_doc(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Extract candidate lessons from documentation text.

    Looks for patterns indicating cause→effect relationships:
      - "if ... then ..." (conditional advice)
      - "to prevent/avoid ... do ..." (preventative advice)
      - "when ... occurs, ..." (troubleshooting)
      - "increase/decrease ... to ..." (parametric advice)
    """
    records = []
    lesson_patterns = [
        # Conditional: "If [condition], [action]"
        (r'(?i)if\s+(.{20,200}?)(?:,\s*|then\s*)(.{20,200}?)(?:\.|$)', "conditional"),
        # Preventative: "To prevent/avoid [problem], [action]"
        (r'(?i)to\s+(?:prevent|avoid|reduce|fix)\s+(.{20,150}?)(?:,\s*|you\s*(?:should|can|need\s*to)?\s*)(.{20,150}?)(?:\.|$)', "preventative"),
        # Troubleshooting: "When/If [symptom], [cause/solution]"
        (r'(?i)(?:when|if)\s+(.{20,150}?)(?:occurs|happens|appears)(?:,\s*|it\s*(?:is|means|indicates)\s*)(.{20,150}?)(?:\.|$)', "troubleshooting"),
        # Parametric: "increase/decrease [param] to [effect]"
        (r'(?i)(increase|decrease|raise|lower)\s+(?:the\s*)?(\w+(?:\s*\w+)?)\s*(?:to|for|when)\s+(.{20,150}?)(?:\.|$)', "parametric"),
    ]

    for pattern, lesson_type in lesson_patterns:
        for match in re.finditer(pattern, text):
            groups = match.groups()
            if lesson_type == "conditional":
                condition, action = groups[0], groups[1]
                lesson_text = f"{condition.strip()} — {action.strip()}."
            elif lesson_type == "preventative":
                problem, action = groups[0], groups[1]
                lesson_text = f"To prevent {problem.strip()}, {action.strip()}."
            elif lesson_type == "troubleshooting":
                symptom, cause = groups[0], groups[1]
                lesson_text = f"When {symptom.strip()} occurs, {cause.strip()}."
            elif lesson_type == "parametric":
                direction, param, effect = groups[0], groups[1], groups[2]
                lesson_text = f"{direction.capitalize()} {param.strip()} to {effect.strip()}."
            else:
                continue

            # Skip if too short or too long
            if len(lesson_text) < 30 or len(lesson_text) > 300:
                continue

            # Try to detect material and geometry from context
            context_start = max(0, match.start() - 300)
            context = text[context_start:match.end()]
            material = _material_of(context) or "PLA"
            geometry = "overhang"  # default
            for gt in GEOMETRY_TYPES:
                if gt in context.lower():
                    geometry = gt
                    break

            # Detect outcome from lesson text
            outcome = "success"
            if any(w in lesson_text.lower() for w in ("fail", "warp", "sag", "string", "crack", "lift", "poor", "bad", "issue", "problem")):
                if "string" in lesson_text.lower():
                    outcome = "failed_stringing"
                else:
                    outcome = "failed_sag"

            job_id = f"modal-doc-{hashlib.md5(lesson_text.encode()).hexdigest()[:8]}"
            records.append(("B", {
                "job_id": job_id,
                "material": material,
                "geometry_type": geometry,
                "env_temp": 22.0,
                "env_humidity": 45.0,
                "outcome": outcome,
                "lesson": lesson_text,
                "source": "ingested",
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }))

    return records


def _extract_pid_values(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Extract PID constants from documentation (specialized parser)."""
    records = []
    # Klipper-style: pid_Kp=22.2 pid_Ki=1.08 pid_Kd=114
    pid_match = re.search(
        r'pid_Kp\s*=\s*([\d.]+).*?pid_Ki\s*=\s*([\d.]+).*?pid_Kd\s*=\s*([\d.]+)',
        text, re.S | re.I
    )
    if pid_match:
        for i, (val, param) in enumerate(zip(pid_match.groups(), ("pid_Kp", "pid_Ki", "pid_Kd"))):
            records.append(("A", {
                "material": "*",
                "param": param,
                "value": float(val),
                "source": source_label,
            }))
    return records


def _extract_linear_advance(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Extract Linear Advance K-factors from documentation."""
    records = []
    # Marlin-style: M900 K0.05 for PLA, K0.08 for PETG
    for match in re.finditer(
        r'(?:M900\s*)?K\s*(\d+(?:\.\d+)?)\s*(?:for\s+)?(\w+)',
        text, re.I
    ):
        try:
            k_value = float(match.group(1))
        except ValueError:
            continue
        # Realistic K-factors are 0.0-2.0 (most filaments 0.0-0.2, flexible up to 2.0)
        if k_value < 0 or k_value > 2.0:
            continue
        material = _material_of(match.group(2)) or match.group(2).upper()
        if material not in MATERIALS:
            material = "*"
        records.append(("A", {
            "material": material,
            "param": "linear_advance_k",
            "value": k_value,
            "source": source_label,
        }))
    return records


def _extract_thermistor_types(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Extract thermistor type → max_temp mappings from Klipper docs."""
    records = []
    # Pattern: "EPCOS 100K B57560G104F" with max_temp nearby
    for match in re.finditer(
        r'(?:thermistor|sensor)_type\s*[:=]\s*[\'"]?(\w+(?:\s+\w+)*)[\'"]?.*?max_temp\s*[:=]\s*(\d+)',
        text, re.S | re.I
    ):
        records.append(("A", {
            "material": "*",
            "param": "max_temp",
            "value": float(match.group(2)),
            "source": f"{source_label}:{match.group(1).strip()}",
        }))
    return records


# ── Source-type dispatchers ───────────────────────────────────────────────────

def _parse_web_doc(text: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse a web documentation page into lane records."""
    records = []
    source_label = f"modal:{source_key}"

    # Clean HTML tags if present
    clean = re.sub(r'<[^>]+>', ' ', text)
    clean = re.sub(r'\s+', ' ', clean).strip()

    if len(clean) < 100:
        return records

    # Run all deterministic parsers
    records.extend(_extract_temperature_values(clean, source_label))
    records.extend(_extract_lessons_from_doc(clean, source_label))
    records.extend(_extract_pid_values(clean, source_label))
    records.extend(_extract_linear_advance(clean, source_label))
    records.extend(_extract_thermistor_types(clean, source_label))

    return records


def _parse_github_repo(repo_path: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse a cloned GitHub repo for config files and documentation."""
    records = []
    repo_dir = Path(repo_path)
    if not repo_dir.exists():
        return records

    source_label = f"modal:{source_key}"

    # Look for config files
    config_patterns = [
        ("**/*.cfg", _parse_klipper_style_cfg),
        ("**/Configuration.h", _parse_marlin_style_h),
        ("**/*.ini", _parse_prusa_style_ini),
        ("**/README.md", _parse_readme),
    ]

    for glob_pattern, parser_fn in config_patterns:
        for filepath in repo_dir.glob(glob_pattern):
            try:
                text = filepath.read_text(encoding="utf-8", errors="ignore")
                records.extend(parser_fn(text, f"{source_label}:{filepath.name}"))
            except Exception:
                continue

    return records


def _parse_prusa_profiles(text: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse PrusaSlicer INI profiles."""
    return _parse_prusa_style_ini(text, f"modal:{source_key}")


def _parse_product_spec(text: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse product specification page for material parameters."""
    records = []
    source_label = f"modal:{source_key}"
    clean = re.sub(r'<[^>]+>', ' ', text)
    clean = re.sub(r'\s+', ' ', clean).strip()

    # Extract temperature ranges
    for match in re.finditer(
        r'(?:print(?:ing)?|nozzle|extruder)\s*temp(?:erature)?\s*(?:range)?\s*[:=]?\s*(\d{3})\s*[-–]\s*(\d{3})\s*°?[CF]?',
        clean, re.I
    ):
        material = _material_of(clean) or "TPU"
        records.append(("A", {
            "material": material,
            "param": "nozzle_temp",
            "value": float(match.group(1)),
            "source": source_label,
        }))

    for match in re.finditer(
        r'(?:bed|heatbed)\s*temp(?:erature)?\s*(?:range)?\s*[:=]?\s*(\d{2,3})\s*[-–]\s*(\d{2,3})\s*°?[CF]?',
        clean, re.I
    ):
        material = _material_of(clean) or "TPU"
        records.append(("A", {
            "material": material,
            "param": "bed_temp",
            "value": float(match.group(1)),
            "source": source_label,
        }))

    # Shore hardness
    shore_match = re.search(r'(?:shore\s*hardness|hardness)\s*[:=]?\s*(\d{2}A)', clean, re.I)
    if shore_match:
        records.append(("A", {
            "material": _material_of(clean) or "TPU",
            "param": "shore_hardness",
            "value": float(shore_match.group(1).replace("A", "")),
            "source": source_label,
        }))

    return records


# ── Config file parsers (reuse logic from distill.py) ─────────────────────────

def _parse_klipper_style_cfg(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Parse Klipper-style cfg for max temps and settings."""
    records = []
    for section, param in (("extruder", "max_temp"), ("heater_bed", "bed_max_temp")):
        m = re.search(rf"\[{section}\][^\[]*?max_temp\s*[:=]\s*(\d+(?:\.\d+)?)", text, re.S | re.I)
        if m:
            records.append(("A", {
                "material": "*",
                "param": param,
                "value": float(m.group(1)),
                "source": source_label,
            }))
    # PID values
    records.extend(_extract_pid_values(text, source_label))
    # Pressure advance
    pa_match = re.search(r'pressure_advance\s*[:=]\s*([\d.]+)', text, re.I)
    if pa_match:
        records.append(("A", {
            "material": "*",
            "param": "pressure_advance",
            "value": float(pa_match.group(1)),
            "source": source_label,
        }))
    return records


def _parse_marlin_style_h(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Parse Marlin Configuration.h for max temps."""
    records = []
    for define, param in (("HEATER_0_MAXTEMP", "max_temp"), ("BED_MAXTEMP", "bed_max_temp")):
        m = re.search(rf"#define\s+{define}\s+(\d+)", text)
        if m:
            records.append(("A", {
                "material": "*",
                "param": param,
                "value": float(m.group(1)),
                "source": source_label,
            }))
    # DEFAULT_Kp/Ki/Kd
    for pid_param in ("DEFAULT_Kp", "DEFAULT_Ki", "DEFAULT_Kd"):
        m = re.search(rf"#define\s+{pid_param}\s+([\d.]+)", text)
        if m:
            records.append(("A", {
                "material": "*",
                "param": pid_param.lower(),
                "value": float(m.group(1)),
                "source": source_label,
            }))
    # Linear advance K factor
    la_match = re.search(r'#define\s+LIN_ADVANCE_K\s+([\d.]+)', text)
    if la_match:
        records.append(("A", {
            "material": "*",
            "param": "linear_advance_k",
            "value": float(la_match.group(1)),
            "source": source_label,
        }))
    return records


def _parse_prusa_style_ini(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Parse PrusaSlicer-style INI for filament settings."""
    records = []
    # Find filament sections and extract key values
    current_material = None
    for line in text.splitlines():
        line = line.strip()
        # Section header: [filament:Generic PLA]
        section_match = re.match(r'\[(?:filament:)?(.+)\]', line, re.I)
        if section_match:
            current_material = _material_of(section_match.group(1))
            continue

        if not current_material:
            continue

        # Key = value pairs
        kv_match = re.match(r'(\w+)\s*=\s*(.+)$', line)
        if not kv_match:
            continue

        key, raw_val = kv_match.group(1), kv_match.group(2).strip()
        num_match = re.search(r'-?\d+(?:\.\d+)?', raw_val.split(",")[0])
        if not num_match:
            continue
        value = float(num_match.group())

        param_map = {
            "temperature": "nozzle_temp",
            "first_layer_temperature": "nozzle_temp",
            "bed_temperature": "bed_temp",
            "first_layer_bed_temperature": "bed_temp",
            "retract_length": "retraction_mm",
            "retract_speed": "retraction_speed",
            "fan_speed": "fan_pct",
            "min_fan_speed": "fan_pct",
            "max_fan_speed": "fan_pct",
        }
        param = param_map.get(key)
        if param:
            records.append(("A", {
                "material": current_material,
                "param": param,
                "value": value,
                "source": source_label,
            }))

    return records


def _parse_readme(text: str, source_label: str) -> list[tuple[str, dict]]:
    """Parse README for reference facts and lessons."""
    records = []
    records.extend(_extract_temperature_values(text, source_label))
    records.extend(_extract_lessons_from_doc(text, source_label))
    return records


# ── Structured profile parsers (Tier S sources) ──────────────────────────────

def _parse_bambu_json_profiles(repo_path: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse BambuStudio tree-structured filament JSON profiles → Lane A.

    Three-layer inheritance: fdm_filament_pla → Material@base → Material@BBL_A1.
    Extracts: nozzle_temp, bed_temp, fan_speed, retraction, volumetric_speed,
    flow_ratio, density, cost per material.
    """
    import json as _json
    records = []
    repo_dir = Path(repo_path)
    filament_dir = repo_dir / "resources" / "profiles" / "BBL" / "filament"
    if not filament_dir.exists():
        return records

    source_label = f"modal:{source_key}"

    # First, load the root defaults (fdm_filament_pla.json, etc.)
    root_defaults = {}
    for root_file in filament_dir.glob("fdm_filament_*.json"):
        try:
            data = _json.loads(root_file.read_text(encoding="utf-8"))
            mat_key = root_file.stem.replace("fdm_filament_", "")
            root_defaults[mat_key] = data
        except Exception:
            continue

    # Then process each material-specific file
    for json_file in sorted(filament_dir.glob("*.json")):
        name = json_file.stem
        # Skip root defaults and non-material files
        if name.startswith("fdm_filament_") or "@" not in name:
            continue

        try:
            data = _json.loads(json_file.read_text(encoding="utf-8"))
        except Exception:
            continue

        # Resolve inheritance chain
        inherits = data.get("inherits", "")
        parent_data = {}
        if inherits in root_defaults:
            parent_data = root_defaults[inherits]

        # Merge: child overrides parent
        merged = {**parent_data, **data}

        # Extract material name from filename: "Bambu PLA Basic @BBL A1" → "PLA"
        material = "*"
        for m in MATERIALS:
            if m in name.upper():
                material = m
                break
        # Handle special materials
        special_map = {
            "PA": "ABS", "PC": "ABS", "ASA": "ABS", "PETG": "PETG",
            "TPU": "TPU", "PLA": "PLA", "ABS": "ABS",
        }
        for key, mat in special_map.items():
            if key in name.upper() and material == "*":
                material = mat
                break

        src = f"{source_label}:{name}"

        # Extract settings (BambuStudio uses array format ["value"])
        def _first_num(val):
            if isinstance(val, list) and val:
                val = val[0]
            if isinstance(val, str):
                try:
                    return float(val.rstrip("%"))
                except ValueError:
                    return None
            if isinstance(val, (int, float)):
                return float(val)
            return None

        param_map = {
            "nozzle_temperature": "nozzle_temp",
            "nozzle_temperature_initial_layer": "nozzle_temp",
            "hot_plate_temp": "bed_temp",
            "hot_plate_temp_initial_layer": "bed_temp",
            "textured_plate_temp": "bed_temp",
            "textured_plate_temp_initial_layer": "bed_temp",
            "fan_max_speed": "fan_pct",
            "fan_min_speed": "fan_pct",
            "filament_max_volumetric_speed": "max_volumetric_speed",
            "filament_flow_ratio": "flow_ratio",
            "filament_density": "density",
            "filament_cost": "cost",
            "filament_retraction_length": "retraction_mm",
            "filament_retraction_speed": "retraction_speed",
            "slow_down_layer_time": "slow_down_layer_time",
        }

        for json_key, param in param_map.items():
            val = _first_num(merged.get(json_key))
            if val is not None:
                records.append(("A", {
                    "material": material,
                    "param": param,
                    "value": val,
                    "source": src,
                }))

    return records


def _parse_klipper_config_repo(repo_path: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse Kanrog Klipper config generator → Lane A.

    150+ .cfg files with motherboard pin maps, PID values, max temps,
    thermistor types, stepper driver settings.
    """
    records = []
    repo_dir = Path(repo_path)
    config_dir = repo_dir / "config-examples"
    if not config_dir.exists():
        return records

    source_label = f"modal:{source_key}"

    for cfg_file in config_dir.glob("*.cfg"):
        try:
            text = cfg_file.read_text(encoding="utf-8", errors="ignore")
        except Exception:
            continue

        src = f"{source_label}:{cfg_file.name}"

        # Reuse existing Klipper parser for max temps + PID
        records.extend(_parse_klipper_style_cfg(text, src))

        # Additional: extract stepper run_current
        for match in re.finditer(r'run_current\s*[:=]\s*([\d.]+)', text):
            records.append(("A", {
                "material": "*",
                "param": "run_current",
                "value": float(match.group(1)),
                "source": src,
            }))

        # Extract thermistor type
        for match in re.finditer(r'sensor_type\s*[:=]\s*(\S+)', text):
            records.append(("A", {
                "material": "*",
                "param": "sensor_type",
                "value": 0,
                "source": f"{src}:{match.group(1)}",
            }))

        # Extract microsteps
        for match in re.finditer(r'microsteps\s*[:=]\s*(\d+)', text):
            records.append(("A", {
                "material": "*",
                "param": "microsteps",
                "value": float(match.group(1)),
                "source": src,
            }))

    return records


def _parse_failure_detection_repo(repo_path: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse 3DPrintSaviour failure detection thresholds → Lane B lessons.

    Extracts the NRMSE threshold constants and classification logic:
      - Detachment: score > 1.0 AND deviance > 1.0
      - Partial breakage: scr_diff > 0.2 AND dev_diff > 0.2
      - Filament runout/clog: score < 0.2 AND deviance < 0.2
      - Spaghetti: ML confidence ≥ 0.3
    """
    records = []
    repo_dir = Path(repo_path)
    source_label = f"modal:{source_key}"
    now = datetime.now(timezone.utc).isoformat()

    # Read printcontrol.py for threshold constants
    pc_path = repo_dir / "printcontrol.py"
    if pc_path.exists():
        text = pc_path.read_text(encoding="utf-8", errors="ignore")

        # Extract threshold constants
        thresholds = {}
        for match in re.finditer(r'(SCR_THRES|DEV_THRES|BR_SCR_THRES|BR_DEV_THRES|FIL_SCR_THRES|FIL_DEV_THRES)\s*=\s*([\d.]+)', text):
            thresholds[match.group(1)] = float(match.group(2))

        # Detachment lesson
        if "SCR_THRES" in thresholds:
            records.append(("B", {
                "job_id": f"modal-saviour-detach-{hashlib.md5(b'detach').hexdigest()[:8]}",
                "material": "PLA",
                "geometry_type": "adhesion",
                "env_temp": 22.0,
                "env_humidity": 45.0,
                "outcome": "failed_sag",
                "lesson": (
                    f"Print detachment detected when layer-to-layer NRMSE score > {thresholds['SCR_THRES']} "
                    f"AND 5-layer deviance > {thresholds.get('DEV_THRES', 1.0)}. "
                    "Large structural changes across consecutive layers indicate the part has separated from the bed. "
                    "Check bed adhesion: clean surface, raise bed temp, use brim or raft."
                ),
                "source": "ingested",
                "timestamp": now,
            }))

        # Clog/runout lesson
        if "FIL_SCR_THRES" in thresholds:
            records.append(("B", {
                "job_id": f"modal-saviour-clog-{hashlib.md5(b'clog').hexdigest()[:8]}",
                "material": "PLA",
                "geometry_type": "stringing",
                "env_temp": 22.0,
                "env_humidity": 45.0,
                "outcome": "failed_stringing",
                "lesson": (
                    f"Filament runout or nozzle clog detected when NRMSE score < {thresholds['FIL_SCR_THRES']} "
                    f"AND deviance < {thresholds.get('FIL_DEV_THRES', 0.2)}. "
                    "Near-zero structural change means no material is being deposited. "
                    "Check filament spool, extruder tension, and nozzle for clogs."
                ),
                "source": "ingested",
                "timestamp": now,
            }))

        # Partial breakage lesson
        if "BR_SCR_THRES" in thresholds:
            records.append(("B", {
                "job_id": f"modal-saviour-break-{hashlib.md5(b'break').hexdigest()[:8]}",
                "material": "PLA",
                "geometry_type": "overhang",
                "env_temp": 22.0,
                "env_humidity": 45.0,
                "outcome": "failed_sag",
                "lesson": (
                    f"Partial breakage detected when frame-to-frame score delta > {thresholds['BR_SCR_THRES']} "
                    f"AND deviance delta > {thresholds.get('BR_DEV_THRES', 0.2)}. "
                    "Sudden structural changes mid-print indicate layer delamination or part fracture. "
                    "Reduce cooling fan, increase nozzle temp, check for drafts."
                ),
                "source": "ingested",
                "timestamp": now,
            }))

    return records


def _parse_filament_profiles_repo(repo_path: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse jklewa filament-profiles-data → Lane A.

    sample-filaments.json: community-verified nozzle/bed temp ranges,
    material type, vendor, price, color per filament SKU.
    """
    import json as _json
    records = []
    repo_dir = Path(repo_path)
    json_path = repo_dir / "sample-filaments.json"
    if not json_path.exists():
        return records

    source_label = f"modal:{source_key}"

    try:
        data = _json.loads(json_path.read_text(encoding="utf-8"))
    except Exception:
        return records

    filaments = data.get("filaments", [])
    for fil in filaments:
        material_name = str(fil.get("material", "")).upper()
        material = "*"
        for m in MATERIALS:
            if m in material_name:
                material = m
                break

        brand = fil.get("brand_name", "unknown")
        color = fil.get("color", "")
        src = f"{source_label}:{brand}/{material_name}/{color}" if color else f"{source_label}:{brand}/{material_name}"

        # Properties: temp_min, temp_max, bed_temp_min, bed_temp_max
        props = fil.get("properties") or fil.get("default_properties") or {}

        for key, param in [("temp_min", "nozzle_temp_min"), ("temp_max", "nozzle_temp_max"),
                            ("bed_temp_min", "bed_temp_min"), ("bed_temp_max", "bed_temp_max")]:
            val = props.get(key)
            if val is not None:
                try:
                    records.append(("A", {"material": material, "param": param,
                                          "value": float(val), "source": src}))
                except (ValueError, TypeError):
                    pass

        # Optional: k_value, flow_ratio, fan_speed_min
        for key, param in [("k_value", "linear_advance_k"), ("flow_ratio", "flow_ratio"),
                            ("fan_speed_min", "fan_pct")]:
            val = props.get(key)
            if val is not None:
                try:
                    records.append(("A", {"material": material, "param": param,
                                          "value": float(val), "source": src}))
                except (ValueError, TypeError):
                    pass

        # Price data
        price_data = fil.get("price_data")
        if price_data and price_data.get("price"):
            try:
                records.append(("A", {"material": material, "param": "price_usd",
                                      "value": float(price_data["price"]), "source": src}))
            except (ValueError, TypeError):
                pass

    return records


def _parse_fdm_error_gcode(repo_path: str, url: str, source_key: str) -> list[tuple[str, dict]]:
    """Parse FDM error detection G-code + YAML → Lane C calibration observations.

    NilsHagenBeyer/3D-printing_recorder: G-code files with known failure outcomes.
    YAML metadata logs: class (GOOD/STRINGING/underextrusion), filament, nozzle,
    retraction, layer_height, extrusion_multiplier.
    G-code filenames encode: temperature, material, printer.

    FIXED (per RESEARCH-NEEDS.md):
    - Parse M106 across the WHOLE file, not just header
    - Parse retraction from G1 E- moves + PrusaSlicer footer comments
    - Infer geometry from G-code moves (bridge/overhang), not bbox heuristic
    - Skip rows where fan_pct can't be determined (no M106 found → null, not 0)
    """
    import yaml as _yaml
    records = []
    repo_dir = Path(repo_path)
    gcode_dir = repo_dir / "gcode"
    if not gcode_dir.exists():
        return records

    source_label = f"modal:{source_key}"

    # Failure class → outcome
    failure_map = {
        "good": "success",
        "stringing": "failed_stringing",
        "underextrusion": "failed_sag",
        "underex": "failed_sag",
    }

    def _infer_geometry_from_gcode(text: str) -> str:
        """Infer geometry type from G-code move patterns.

        Bridge: long X/Y moves at the same Z with extrusion → spanning gaps.
        Overhang: outward-stepping perimeters (X/Y expanding each layer).
        Stringing: many travel moves (G0) between disconnected regions.
        Default: overhang.
        """
        # Count travel moves vs extrusion moves
        travels = len(re.findall(r'\nG0\s', text))
        extrudes = len(re.findall(r'\nG1\s.*?E', text))
        if extrudes > 0 and travels > extrudes * 0.3:
            return "stringing"  # high travel ratio = disconnected regions

        # Look for bridge patterns: long X/Y moves at same Z
        bridge_moves = 0
        prev_z = None
        for match in re.finditer(r'G1\s.*?X([\d.]+)\s+Y([\d.]+)\s+Z([\d.]+).*?E([\d.]+)', text):
            z = float(match.group(3))
            e = float(match.group(4))
            if prev_z is not None and abs(z - prev_z) < 0.01 and e > 0.5:
                dx = abs(float(match.group(1)) - prev_x) if 'prev_x' in dir() else 0
                dy = abs(float(match.group(2)) - prev_y) if 'prev_y' in dir() else 0
                if dx > 30 or dy > 30:
                    bridge_moves += 1
            prev_z = z
            prev_x = float(match.group(1))
            prev_y = float(match.group(2))
        if bridge_moves > 3:
            return "bridge"

        return "overhang"

    def _parse_retraction(text: str, yaml_entry: dict) -> float | None:
        """Parse retraction from G-code moves + slicer footer + YAML."""
        # 1. YAML metadata (most reliable)
        yaml_ret = yaml_entry.get("retraction")
        if yaml_ret is not None:
            return float(yaml_ret)

        # 2. PrusaSlicer/Orca footer comment block (last 2KB of file)
        footer = text[-2000:] if len(text) > 2000 else text
        footer_match = re.search(r';\s*retract_length\s*=\s*([\d.]+)', footer, re.I)
        if footer_match:
            return float(footer_match.group(1))

        # 3. G1 E- retraction moves (negative extrusion = retract)
        retract_moves = re.findall(r'G1\s.*?E-([\d.]+)', text)
        if retract_moves:
            return float(retract_moves[0])

        return None

    def _parse_fan(text: str) -> float | None:
        """Parse fan speed from M106 across the WHOLE file + slicer footer."""
        # 1. M106 commands anywhere in the file
        fan_matches = re.findall(r'M106\s+S(\d+)', text)
        if fan_matches:
            # Use the most common non-zero fan speed
            fans = [int(f) for f in fan_matches if int(f) > 0]
            if fans:
                # Convert 0-255 PWM to 0-100%
                fan_val = max(set(fans), key=fans.count)
                return round(fan_val / 255.0 * 100, 1)

        # 2. PrusaSlicer/Orca footer comment block
        footer = text[-3000:] if len(text) > 3000 else text
        for key in ['fan_speed', 'fan_percentage', 'cooling_fan_speed', 'bridge_fan_speed']:
            m = re.search(rf';\s*{key}\s*=\s*([\d.]+)', footer, re.I)
            if m:
                return float(m.group(1))

        # 3. M107 (fan off) — explicit off is valid data
        if re.search(r'M107', text):
            return 0.0

        return None  # Can't determine — skip this row

    # Process each macro run directory
    for macro_dir in sorted(gcode_dir.iterdir()):
        if not macro_dir.is_dir():
            continue

        # Determine failure type from directory name
        dir_name = macro_dir.name.lower()
        failure_type = "good"
        if "stringing" in dir_name:
            failure_type = "stringing"
        elif "underex" in dir_name:
            failure_type = "underextrusion"
        elif "good" in dir_name:
            failure_type = "good"

        outcome = failure_map.get(failure_type, "success")

        # Load YAML metadata if present
        yaml_data = {}
        for yaml_file in macro_dir.glob("*.yaml"):
            try:
                yaml_data = _yaml.safe_load(yaml_file.read_text(encoding="utf-8"))
            except Exception:
                pass

        # Build lookup from YAML: gcode filename → metadata
        yaml_lookup = {}
        if isinstance(yaml_data, list):
            for entry in yaml_data:
                gcode_name = entry.get("gcode", "")
                if gcode_name:
                    yaml_lookup[gcode_name] = entry

        # Process each G-code file
        for gcode_file in sorted(macro_dir.glob("*.gcode")):
            try:
                full_text = gcode_file.read_text(encoding="utf-8", errors="ignore")
            except Exception:
                continue

            # Parse settings from the FULL file — use FINAL temps (M109/M190), not preheat
            nozzle_matches = re.findall(r'M109\s+S(\d+)', full_text)  # wait-for-nozzle = final temp
            if not nozzle_matches:
                nozzle_matches = re.findall(r'M104\s+S(\d+)', full_text)
                if nozzle_matches:
                    nozzle_temp = float(nozzle_matches[-1])  # last M104 = final
                else:
                    continue
            else:
                nozzle_temp = float(nozzle_matches[-1])

            bed_matches = re.findall(r'M190\s+S(\d+)', full_text)  # wait-for-bed = final temp
            if not bed_matches:
                bed_matches = re.findall(r'M140\s+S(\d+)', full_text)
            bed_temp = float(bed_matches[-1]) if bed_matches else 60.0

            # Parse fan — skip row if can't determine (RESEARCH-NEEDS.md fix #1)
            fan_pct = _parse_fan(full_text)
            if fan_pct is None:
                continue  # Skip — can't trust fan=0 default

            # Parse filename for material
            fname = gcode_file.name.upper()
            material = "PLA"
            for m in MATERIALS:
                if m in fname:
                    material = m
                    break

            # Get YAML metadata
            yaml_entry = yaml_lookup.get(gcode_file.name, {})

            # Parse retraction (RESEARCH-NEEDS.md fix #2)
            retraction = _parse_retraction(full_text, yaml_entry)
            if retraction is None:
                retraction = 5.0  # fallback default

            # Infer geometry from G-code moves (RESEARCH-NEEDS.md fix #3)
            geometry_type = _infer_geometry_from_gcode(full_text)
            # Override with failure type if it's more specific
            if failure_type == "stringing":
                geometry_type = "stringing"

            # Quality estimate from extrusion multiplier
            ex_mul = float(yaml_entry.get("extrusion_multiplier", yaml_entry.get("ex_mul", 1.0)))
            quality = max(0.1, min(1.0, ex_mul)) if failure_type == "good" else max(0.1, min(0.7, 1.0 - abs(1.0 - ex_mul)))

            records.append(("C", {
                "material": material,
                "geometry_type": geometry_type,
                "env_temp": 22.0,
                "env_humidity": 45.0,
                "nozzle_temp": nozzle_temp,
                "bed_temp": bed_temp,
                "retraction_mm": retraction,
                "fan_pct": fan_pct,
                "first_layer_fan_pct": 0,
                "outcome": outcome,
                "quality": round(quality, 2),
            }))

    return records


# ── Source-type → parser dispatch ─────────────────────────────────────────────

PARSER_DISPATCH = {
    "web_doc": _parse_web_doc,
    "github_repo": _parse_github_repo,
    "prusa_profiles": _parse_prusa_profiles,
    "product_spec": _parse_product_spec,
    "research_repo": _parse_github_repo,
    "bambu_json_profiles": _parse_bambu_json_profiles,
    "klipper_config_repo": _parse_klipper_config_repo,
    "failure_detection_repo": _parse_failure_detection_repo,
    "filament_profiles_repo": _parse_filament_profiles_repo,
    "fdm_error_gcode": _parse_fdm_error_gcode,
}


# ═══════════════════════════════════════════════════════════════════════════════
# Modal App
# ═══════════════════════════════════════════════════════════════════════════════

if modal is not None:
    app = modal.App("chief-engineer-ingest")

    # Persistent volume for caching fetched content (Spanish tutor pattern)
    vol = modal.Volume.from_name("chief-engineer-ingest-data", create_if_missing=True)

    # Image with dependencies
    image = (
        modal.Image.debian_slim()
        .pip_install(
            "datasets", "pyarrow", "huggingface_hub",
            "pydantic>=2.7", "requests", "beautifulsoup4",
            "pyyaml",
        )
        .apt_install("git", "wget")
        # Share enums into the image (Spanish tutor pattern)
        # Must be last — add_local_* after build steps
        .env({"PYTHONPATH": "/root"})
        .add_local_file("core/models.py", "/root/core/models.py")
    )

    # ── HF Dataset Distillation ───────────────────────────────────────────

    @app.function(
        image=image,
        timeout=3600,
        cpu=4,
        memory=4096,
        secrets=[modal.Secret.from_name("chief-engineer-secrets")],
    )
    def distill_hf_dataset(dataset: str, limit: int = 1000, split: str = "train") -> dict:
        """Load an HF dataset, map rows → lane records.

        Special case: 3dtime loads from metadata CSV (not standard rows).
        """
        import csv
        import io
        import requests

        if dataset not in HF_MAPPERS:
            return {"error": f"unknown dataset '{dataset}'; known: {list(HF_MAPPERS)}"}
        hf_id, mapper = HF_MAPPERS[dataset]

        out: dict[str, list] = {"A": [], "B": [], "C": []}

        # 3DTime special case: download metadata CSV + G-code headers
        if dataset == "3dtime":
            csv_url = f"https://huggingface.co/datasets/{hf_id}/resolve/main/metadata/metadata_sub21.csv"
            resp = requests.get(csv_url, timeout=30)
            resp.raise_for_status()
            reader = csv.DictReader(io.StringIO(resp.text))
            rows = list(reader)[:limit]

            for row in rows:
                # Lane A: metadata reference facts
                for lane, rec in mapper(row):
                    if lane in _VALID_LANES:
                        out[lane].append(rec)

                # Lane C: download G-code header + footer, extract settings
                gcode_name = row.get("G-code file name", "")
                if gcode_name:
                    try:
                        gcode_url = f"https://huggingface.co/datasets/{hf_id}/resolve/main/sliced/21/{gcode_name}"
                        # Download header (first 10KB) for nozzle/bed temps
                        gcode_resp = requests.get(gcode_url, stream=True, timeout=15)
                        header = ""
                        for chunk in gcode_resp.iter_content(chunk_size=1024):
                            header += chunk.decode("utf-8", errors="ignore")
                            if len(header) > 10000:
                                break
                        # Also try to get footer (last 3KB) for fan/retraction settings
                        footer = ""
                        try:
                            content_length = gcode_resp.headers.get("Content-Length")
                            if content_length:
                                size = int(content_length)
                                if size > 3000:
                                    footer_resp = requests.get(gcode_url, headers={"Range": f"bytes={size-3000}-{size}"}, timeout=10)
                                    footer = footer_resp.text
                        except Exception:
                            pass
                        full_text = header + footer

                        # Parse nozzle/bed — use FINAL temps (M109/M190)
                        nozzle_matches = re.findall(r'M109\s+S(\d+)', full_text)
                        if not nozzle_matches:
                            nozzle_matches = re.findall(r'M104\s+S(\d+)', full_text)
                        bed_matches = re.findall(r'M190\s+S(\d+)', full_text)
                        if not bed_matches:
                            bed_matches = re.findall(r'M140\s+S(\d+)', full_text)
                        if not nozzle_matches or not bed_matches:
                            continue
                        nozzle_temp = float(nozzle_matches[-1])
                        bed_temp = float(bed_matches[-1])

                        # Parse fan from M106 across full text + slicer footer
                        fan_matches = re.findall(r'M106\s+S(\d+)', full_text)
                        fan_pct = None
                        if fan_matches:
                            fans = [int(f) for f in fan_matches if int(f) > 0]
                            if fans:
                                fan_pct = round(max(set(fans), key=fans.count) / 255.0 * 100, 1)
                        if fan_pct is None:
                            # Check slicer footer comments
                            for key in ['fan_speed', 'fan_percentage', 'cooling_fan_speed', 'bridge_fan_speed']:
                                m = re.search(rf';\s*{key}\s*=\s*([\d.]+)', full_text, re.I)
                                if m:
                                    fan_pct = float(m.group(1))
                                    break
                        if fan_pct is None and re.search(r'M107', full_text):
                            fan_pct = 0.0  # explicit fan off
                        if fan_pct is None:
                            continue  # Skip — can't determine fan (RESEARCH-NEEDS.md fix)

                        material_map = {"pla": "PLA", "pet": "PETG", "abs": "ABS", "tpu": "TPU"}
                        material = material_map.get(str(row.get("Material", "")).lower(), "PLA")

                        # Parse retraction from footer or G-code
                        retraction = 5.0
                        footer_ret = re.search(r';\s*retract_length\s*=\s*([\d.]+)', full_text, re.I)
                        if footer_ret:
                            retraction = float(footer_ret.group(1))
                        else:
                            retract_moves = re.findall(r'G1\s.*?E-([\d.]+)', full_text)
                            if retract_moves:
                                retraction = float(retract_moves[0])

                        # Geometry from bounding box (3DTime has no G-code move context in header)
                        try:
                            bx, by, bz = float(row["Bounding box X (mm)"]), float(row["Bounding box Y (mm)"]), float(row["Bounding box Z (mm)"])
                        except (KeyError, ValueError):
                            bx, by, bz = 50, 50, 20
                        ratio = bx / max(by, 0.1)
                        if bz < 5:
                            geometry = "vase"
                        elif ratio > 5:
                            geometry = "bridge"
                        elif bz > bx * 0.8:
                            geometry = "adhesion"
                        else:
                            geometry = "overhang"

                        out["C"].append({
                            "material": material,
                            "geometry_type": geometry,
                            "env_temp": 22.0,
                            "env_humidity": 45.0,
                            "nozzle_temp": float(nozzle_matches[-1]),
                            "bed_temp": float(bed_matches[-1]),
                            "retraction_mm": retraction,
                            "fan_pct": fan_pct,
                            "first_layer_fan_pct": 0,
                            "outcome": "success",
                            "quality": 0.85,
                        })
                    except Exception:
                        pass
            out["stats"] = {
                "dataset": dataset, "hf_id": hf_id, "rows": len(rows),
                "A": len(out["A"]), "B": len(out["B"]), "C": len(out["C"]),
            }
            return out

        # Standard HF dataset
        from datasets import load_dataset
        ds = load_dataset(hf_id, split=f"{split}[:{limit}]")
        for row in ds:
            for lane, rec in mapper(dict(row)):
                if lane in _VALID_LANES:
                    out[lane].append(rec)
        out["stats"] = {
            "dataset": dataset, "hf_id": hf_id, "rows": len(ds),
            "A": len(out["A"]), "B": len(out["B"]), "C": len(out["C"]),
        }
        return out

    # ── Documentation Source Fetching ─────────────────────────────────────

    @app.function(
        image=image,
        volumes={"/data": vol},
        timeout=600,
        secrets=[modal.Secret.from_name("chief-engineer-secrets")],
    )
    def fetch_and_parse_source(source_key: str) -> dict:
        """Fetch a documentation source, parse it, return lane records.

        Uses Modal Volume for caching: if already fetched, skips download.
        Idempotent/resumable (Spanish tutor pattern).
        """
        import requests
        import subprocess
        import tempfile

        if source_key not in SOURCES:
            return {"error": f"unknown source '{source_key}'", "source_key": source_key}

        url, category, source_type, description = SOURCES[source_key]
        cache_dir = Path("/data/cache")
        cache_dir.mkdir(parents=True, exist_ok=True)

        # Cache key from URL hash
        cache_key = hashlib.md5(url.encode()).hexdigest()[:12]
        content_path = cache_dir / f"{source_key}_{cache_key}.txt"
        records_path = cache_dir / f"{source_key}_{cache_key}_records.json"

        # Check cache — idempotent/resumable
        if records_path.exists():
            try:
                cached = json.loads(records_path.read_text())
                print(f"  ✓ {source_key}: loaded from cache ({cached.get('stats', {}).get('total', 0)} records)")
                return cached
            except Exception:
                pass

        # Fetch content
        content = ""
        if source_type == "github_repo" or source_type == "research_repo" or source_type in ("bambu_json_profiles", "klipper_config_repo", "failure_detection_repo", "filament_profiles_repo", "fdm_error_gcode"):
            # Download repo as zip (more reliable than git clone on Modal)
            repo_name = url.rstrip("/").split("/")[-1]
            repo_dir = cache_dir / f"repo_{source_key}_{cache_key}"
            if not repo_dir.exists():
                print(f"  downloading {url} → {repo_dir}")
                zip_url = f"https://github.com/{url.split('github.com/')[-1]}/archive/refs/heads/main.zip"
                try:
                    resp = requests.get(zip_url, timeout=120, headers={"User-Agent": "chief-engineer/1.0"})
                    resp.raise_for_status()
                    import zipfile as _zipfile
                    import io as _io
                    with _zipfile.ZipFile(_io.BytesIO(resp.content)) as zf:
                        # Extract all files, stripping the top-level directory
                        for member in zf.namelist():
                            # Strip the leading repo-name-branch directory
                            parts = member.split("/", 1)
                            if len(parts) > 1:
                                target = repo_dir / parts[1]
                                if member.endswith("/"):
                                    target.mkdir(parents=True, exist_ok=True)
                                else:
                                    target.parent.mkdir(parents=True, exist_ok=True)
                                    with zf.open(member) as src, open(target, "wb") as dst:
                                        dst.write(src.read())
                    print(f"  ✓ extracted to {repo_dir}")
                except Exception as e:
                    print(f"  ⚠ download failed: {e}")
                    return {"error": f"download failed: {e}", "source_key": source_key}

            # Parse the cloned repo
            records = _parse_github_repo(str(repo_dir), url, source_key)
            content = f"[cloned repo at {repo_dir}]"

        elif source_type == "prusa_profiles":
            # Download INI files from PrusaSlicer repository
            print(f"  fetching Prusa profiles from {url}")
            try:
                resp = requests.get(url, timeout=30, headers={"User-Agent": "chief-engineer/1.0"})
                resp.raise_for_status()
                content = resp.text
                # Parse INI content
                records = _parse_prusa_profiles(content, url, source_key)
            except Exception as e:
                print(f"  ⚠ fetch failed: {e}")
                return {"error": f"fetch failed: {e}", "source_key": source_key}

        else:
            # web_doc, product_spec — fetch HTML
            print(f"  fetching {url}")
            try:
                resp = requests.get(url, timeout=30, headers={"User-Agent": "chief-engineer/1.0"})
                resp.raise_for_status()
                content = resp.text
            except Exception as e:
                print(f"  ⚠ fetch failed: {e}")
                return {"error": f"fetch failed: {e}", "source_key": source_key}

            # Parse based on source type
            parser = PARSER_DISPATCH.get(source_type, _parse_web_doc)
            records = parser(content, url, source_key)

        # Save content cache
        content_path.write_text(content[:50000], encoding="utf-8")  # truncate for cache

        # Build result
        out: dict[str, list] = {"A": [], "B": [], "C": []}
        for lane, rec in records:
            if lane in _VALID_LANES:
                out[lane].append(rec)

        stats = {
            "source_key": source_key,
            "category": category,
            "url": url,
            "A": len(out["A"]),
            "B": len(out["B"]),
            "C": len(out["C"]),
            "total": len(records),
        }
        out["stats"] = stats

        # Cache records
        records_path.write_text(json.dumps(out))

        # Commit volume (Spanish tutor pattern)
        vol.commit()

        print(f"  ✓ {source_key}: {stats['total']} records (A:{stats['A']} B:{stats['B']} C:{stats['C']})")
        return out

    # ── Main Entrypoint ───────────────────────────────────────────────────

    @app.local_entrypoint()
    def main(
        source: str = "",
        category: str = "",
        dataset: str = "",
        limit: int = 1000,
    ):
        """Run locally: fan out to Modal, then write artifacts.

        Usage:
            modal run ingest/modal_app.py                          # all sources
            modal run ingest/modal_app.py --source klipper-config  # single source
            modal run ingest/modal_app.py --category calibration   # category
            modal run ingest/modal_app.py --dataset 3d-adam       # HF dataset
        """
        root = Path(__file__).resolve().parent.parent
        agg: dict[str, list] = {"A": [], "B": [], "C": []}

        # ── Process documentation sources ──
        if not dataset:
            targets = []
            if source:
                if source not in SOURCES:
                    print(f"Unknown source: {source}")
                    print(f"Known sources: {list(SOURCES.keys())}")
                    return
                targets = [source]
            elif category:
                targets = [k for k, v in SOURCES.items() if v[1] == category]
                if not targets:
                    print(f"No sources in category '{category}'")
                    print(f"Categories: {set(v[1] for v in SOURCES.values())}")
                    return
            else:
                targets = list(SOURCES.keys())

            print(f"Processing {len(targets)} documentation sources...\n")
            for src_key in targets:
                res = fetch_and_parse_source.remote(src_key)
                stats = res.get("stats", res)
                if "error" in stats:
                    print(f"  ✗ {src_key}: {stats['error']}")
                else:
                    print(f"  ✓ {src_key}: A={stats.get('A',0)} B={stats.get('B',0)} C={stats.get('C',0)}")
                for lane in ("A", "B", "C"):
                    agg[lane].extend(res.get(lane, []))

        # ── Process HF datasets ──
        if dataset:
            targets = [dataset] if dataset else list(HF_MAPPERS)
            print(f"Processing {len(targets)} HF datasets...\n")
            for d in targets:
                res = distill_hf_dataset.remote(d, limit)
                stats = res.get("stats", res)
                print(f"  {stats}")
                for lane in ("A", "B", "C"):
                    agg[lane].extend(res.get(lane, []))

        # ── Write artifacts ──
        def _write(path: Path, rows: list, append: bool):
            if not rows:
                return
            path.parent.mkdir(parents=True, exist_ok=True)
            with path.open("a" if append else "w") as f:
                for r in rows:
                    f.write(json.dumps(r) + "\n")
            print(f"  wrote {len(rows)} → {path.relative_to(root)}")

        print(f"\n── Artifacts ──")
        _write(root / "data" / "references.jsonl", agg["A"], append=True)
        _write(root / "data" / "_modal_candidate_lessons.jsonl", agg["B"], append=True)
        _write(root / "sim" / "calibration" / "observations.modal.jsonl", agg["C"], append=True)

        print(f"\n── Summary ──")
        print(f"  Lane A (references):     {len(agg['A'])}")
        print(f"  Lane B (candidate lessons): {len(agg['B'])}  ← REVIEW before ledger!")
        print(f"  Lane C (calibration obs):  {len(agg['C'])}")
        print(f"\nNext:")
        print(f"  • REVIEW data/_modal_candidate_lessons.jsonl (honesty gate)")
        print(f"  • Fold good lessons into ledger via ingest_candidate_lessons")
        print(f"  • Calibrate: uv run python -m sim.calibrate --data sim/calibration/observations.modal.jsonl")
        print(f"  • make test  →  make run")

    # FRONTIER (named in writeup, NOT in-window):
    # Fine-tune a small Gemma on the accumulated ledger.
    # @app.function(gpu="A10G", image=image, timeout=3600)
    # def finetune_on_ledger(...): ...


if __name__ == "__main__":
    print("Modal ingestion app for Chief Engineer.")
    print(f"  {len(SOURCES)} documentation sources registered")
    print(f"  {len(HF_MAPPERS)} HF dataset mappers registered")
    print()
    print("Categories:")
    for cat in sorted(set(v[1] for v in SOURCES.values())):
        count = sum(1 for v in SOURCES.values() if v[1] == cat)
        print(f"  {cat}: {count} sources")
    print()
    print("Run:")
    print("  modal run ingest/modal_app.py                          # all sources")
    print("  modal run ingest/modal_app.py --source klipper-thermistors")
    print("  modal run ingest/modal_app.py --category calibration")
    print("  modal run ingest/modal_app.py --dataset 3d-adam --limit 2000")