scripts/audit_converted_solutions.py

#!/usr/bin/env python3
"""Audit converted JSONL files for code in solution fields."""

from __future__ import annotations

import json
from pathlib import Path

from sft_pipeline_utils import asks_for_code, has_code, is_weak_solution

ROOT = Path(__file__).resolve().parent.parent
CONVERTED_DIR = ROOT / "data" / "converted"
REPORT_PATH = ROOT / "data" / "audit" / "converted_solution_audit.md"


def load_all_converted() -> list[dict]:
    rows: list[dict] = []
    for path in sorted(CONVERTED_DIR.glob("*.jsonl")):
        with path.open("r", encoding="utf-8") as f:
            for line in f:
                if line.strip():
                    row = json.loads(line)
                    row["_source_file"] = path.name
                    rows.append(row)
    return rows


def is_planning_only(solution: str) -> bool:
    if has_code(solution):
        return False
    s = str(solution or "").strip()
    if not s:
        return True
    if s.startswith("1)") or s.startswith("- "):
        return True
    if is_weak_solution(s):
        return True
    return False


def main() -> int:
    rows = load_all_converted()
    with_code = [r for r in rows if has_code(r.get("solution", ""))]
    planning_only = [r for r in rows if is_planning_only(r.get("solution", ""))]
    code_ask_no_code = [
        r for r in rows if asks_for_code(r.get("user_prompt", "")) and not has_code(r.get("solution", ""))
    ]

    bad = sorted(code_ask_no_code + [r for r in planning_only if r not in code_ask_no_code], key=lambda r: r.get("id", ""))[:10]
    good = sorted(with_code, key=lambda r: len(str(r.get("solution", ""))), reverse=True)[:10]

    by_file: dict[str, int] = {}
    code_by_file: dict[str, int] = {}
    for r in rows:
        f = r.get("_source_file", "?")
        by_file[f] = by_file.get(f, 0) + 1
        if has_code(r.get("solution", "")):
            code_by_file[f] = code_by_file.get(f, 0) + 1

    lines = [
        "# Converted Solution Audit",
        "",
        f"Source: `{CONVERTED_DIR}`",
        "",
        "## Summary",
        "",
        f"- Total converted rows: **{len(rows)}**",
        f"- Rows with real code in solution: **{len(with_code)}** ({100*len(with_code)/max(len(rows),1):.1f}%)",
        f"- Rows with planning-only solution: **{len(planning_only)}** ({100*len(planning_only)/max(len(rows),1):.1f}%)",
        f"- Code-request prompts without code in solution: **{len(code_ask_no_code)}**",
        "",
        "## Per file",
        "",
        "| File | Rows | With code |",
        "| --- | ---: | ---: |",
    ]
    for f in sorted(by_file):
        lines.append(f"| `{f}` | {by_file[f]} | {code_by_file.get(f, 0)} |")

    lines.extend(["", "## Root cause signals", ""])
    lines.append(
        "- Rows converted **without** raw `solution_code` get numbered `agent_behavior` steps in `solution` (no code)."
    )
    lines.append(
        "- `build_failure_log()` copies the full `instruction` into `failure_log`, which later becomes Diagnosis text."
    )
    lines.append(
        "- Only `code_output_correction_converted.jsonl` consistently has code; game/restaurant batches do not."
    )

    lines.extend(["", "## 10 bad examples (code request or planning-only)", ""])
    for r in bad[:10]:
        lines.append(f"### `{r.get('id', '?')}` ({r.get('_source_file', '?')})")
        lines.append(f"- **user_prompt:** {r.get('user_prompt', '')[:200]}")
        lines.append(f"- **solution (first 300 chars):** {str(r.get('solution', ''))[:300]}")
        lines.append("")

    lines.extend(["", "## 10 good examples (code in solution)", ""])
    for r in good[:10]:
        lines.append(f"### `{r.get('id', '?')}` ({r.get('_source_file', '?')})")
        lines.append(f"- **user_prompt:** {r.get('user_prompt', '')[:200]}")
        lines.append(f"- **solution (first 400 chars):** {str(r.get('solution', ''))[:400]}")
        lines.append("")

    REPORT_PATH.parent.mkdir(parents=True, exist_ok=True)
    REPORT_PATH.write_text("\n".join(lines), encoding="utf-8")
    print(f"Wrote {REPORT_PATH}")
    print(f"Total: {len(rows)}, with code: {len(with_code)}, code-ask-no-code: {len(code_ask_no_code)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())