Update pipeline.py

pipeline.py

@@ -4,9 +4,10 @@ import os
 import re
 import hashlib
 import shutil
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import fitz  # pymupdf
 import pytesseract
@@ -15,34 +16,12 @@ from openai import OpenAI
 from tenacity import retry, stop_after_attempt, wait_exponential
 
 
-# =========================
-# Constants / Contract
-# =========================
 SCHEMA_VERSION = "1.0"
 DEFAULT_MODEL = "gpt-4o-mini"
 
 ALLOWED_SCORE_KEYS = ["skill", "experience", "growth", "context_fit", "combined"]
 
-# New job folder layout (stable for future FastAPI/worker/SFTP)
-JOBS_DIRNAME = "jobs"
 
-INPUT_DIRNAME = "input"
-TEXT_DIRNAME = "extracted_text"
-EVAL_DIRNAME = "evaluations"
-REPORTS_DIRNAME = "reports"
-
-JOB_JSON_NAME = "job.json"
-JOB_INDEX_NAME = "resumes_index.json"
-ARTIFACTS_JSON_NAME = "artifacts.json"
-
-# Global persistent state (idempotency across runs)
-GLOBAL_REPORTS_DIRNAME = "reports"
-GLOBAL_MANIFEST_NAME = "processed_manifest.json"
-
-
-# =========================
-# Helpers
-# =========================
 def _now_ts() -> str:
     return datetime.now(timezone.utc).isoformat()
 
@@ -62,48 +41,37 @@ def _sha256_file(path: str) -> str:
     return h.hexdigest()
 
 
-def _atomic_write_json(path: Path, obj: Any) -> None:
+def _atomic_write_text(path: Path, text: str) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     tmp = path.with_suffix(path.suffix + ".tmp")
-    tmp.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
+    tmp.write_text(text, encoding="utf-8")
     tmp.replace(path)
 
 
+def _atomic_write_json(path: Path, obj: Any) -> None:
+    _atomic_write_text(path, json.dumps(obj, ensure_ascii=False, indent=2))
+
+
 def _read_json(path: Path, default: Any) -> Any:
     if not path.exists():
         return default
     try:
         return json.loads(path.read_text(encoding="utf-8"))
     except Exception:
-        backup = path.with_suffix(path.suffix + ".corrupt.json")
+        # keep a backup of corrupt state and start fresh
         try:
-            shutil.copy2(path, backup)
+            shutil.copy2(path, path.with_suffix(path.suffix + ".corrupt"))
         except Exception:
             pass
         return default
 
 
-def _coerce_score(v: Any) -> float:
-    try:
-        f = float(v)
-    except Exception:
-        return 0.0
-    if f < 0:
-        return 0.0
-    if f > 10:
-        return 10.0
-    return f
-
-
 def _pixmap_to_pil_rgb(pix: "fitz.Pixmap") -> Image.Image:
     if pix.alpha:
         pix = fitz.Pixmap(pix, 0)
     return Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
 
 
-# =========================
-# Text extraction (PyMuPDF + OCR fallback)
-# =========================
 def extract_text_from_pdf(
     pdf_path: str,
     *,
@@ -123,7 +91,7 @@ def extract_text_from_pdf(
     parts: List[str] = []
     page_count = min(len(doc), max_pages)
 
-    # Normal extraction
+    # normal extraction
     for i in range(page_count):
         try:
             t = doc[i].get_text("text") or ""
@@ -154,9 +122,6 @@ def extract_text_from_pdf(
     return "\n\n".join(ocr_parts).strip()
 
 
-# =========================
-# LLM evaluation
-# =========================
 def build_prompt(text: str, config: Dict[str, Any]) -> str:
     projects = config.get("projects") or []
     projects_block = json.dumps(projects, ensure_ascii=False)
@@ -200,7 +165,19 @@ Resume text:
 """.strip()
 
 
-def normalize_eval(raw: Dict[str, Any], config: Dict[str, Any]) -> Dict[str, Any]:
+def _coerce_score(v: Any) -> float:
+    try:
+        f = float(v)
+    except Exception:
+        return 0.0
+    if f < 0:
+        return 0.0
+    if f > 10:
+        return 10.0
+    return f
+
+
+def normalize_eval(raw: Dict[str, Any], config: Dict[str, Any], *, job_id: str, pdf_sha256: str, filename: str) -> Dict[str, Any]:
     scores = raw.get("scores") if isinstance(raw.get("scores"), dict) else {}
     norm_scores = {k: _coerce_score(scores.get(k, 0)) for k in ALLOWED_SCORE_KEYS}
 
@@ -226,6 +203,9 @@ def normalize_eval(raw: Dict[str, Any], config: Dict[str, Any]) -> Dict[str, Any
 
     return {
         "schema_version": SCHEMA_VERSION,
+        "job_id": job_id,
+        "pdf_sha256": pdf_sha256,
+        "filename": filename,
         "candidate_name": raw.get("candidate_name"),
         "seniority": raw.get("seniority"),
         "scores": norm_scores,
@@ -254,9 +234,7 @@ def llm_evaluate(text: str, config: Dict[str, Any]) -> Dict[str, Any]:
     try:
         raw = json.loads(content)
     except Exception as e:
-        raise RuntimeError(
-            f"LLM did not return valid JSON. First 200 chars: {content[:200]!r}"
-        ) from e
+        raise RuntimeError(f"LLM did not return valid JSON. First 200 chars: {content[:200]!r}") from e
 
     if not isinstance(raw, dict):
         raise RuntimeError("LLM JSON must be an object/dict at top-level.")
@@ -264,190 +242,130 @@ def llm_evaluate(text: str, config: Dict[str, Any]) -> Dict[str, Any]:
     return raw
 
 
-# =========================
-# Records / Derived Reports
-# =========================
-def _make_record_base(pdf_path: str, config: Dict[str, Any], project_name: str, sha: str) -> Dict[str, Any]:
-    filename = os.path.basename(pdf_path)
-    model = config.get("model") or os.getenv("OPENAI_MODEL") or DEFAULT_MODEL
-    return {
-        "schema_version": SCHEMA_VERSION,
-        "pdf_sha256": sha,
-        "filename": filename,
-        "candidate_name": None,
-        "project": project_name,
-        "model": model,
-        "status": None,          # success|skipped|failed
-        "error": None,
-        "created_at": _now_ts(),
-        "output_json": None,     # relative under job_dir
-        "extracted_text": None,  # relative under job_dir
-    }
-
-
-def _bucket_for_score(score: float) -> str:
-    # Adjust thresholds as needed; keep deterministic.
-    if score >= 8.0:
+def _bucket_label(combined: float, thresholds: Dict[str, float]) -> str:
+    top = float(thresholds.get("top", 8.0))
+    strong = float(thresholds.get("strong", 6.5))
+    maybe = float(thresholds.get("maybe", 5.0))
+    if combined >= top:
         return "top"
-    if score >= 6.5:
+    if combined >= strong:
         return "strong"
-    if score >= 5.0:
+    if combined >= maybe:
         return "maybe"
     return "no"
 
 
-def _compute_reports(job_index: List[Dict[str, Any]]) -> Dict[str, Any]:
-    """
-    Derive:
-      - project_buckets.{json,csv}
-      - top_candidates.{json,csv}
-      - candidate_analysis.csv
-    from successful evaluations only.
-    """
-    rows = [r for r in job_index if r.get("status") == "success"]
-
-    # project buckets (by "combined")
-    buckets: Dict[str, Dict[str, List[Dict[str, Any]]]] = {}
-    for r in rows:
-        project = r.get("project") or "UNKNOWN"
-        combined = r.get("combined_score", 0.0)
-        b = _bucket_for_score(float(combined or 0.0))
-        buckets.setdefault(project, {}).setdefault(b, []).append(r)
-
-    # top candidates: sort by combined desc
-    ranked = sorted(rows, key=lambda x: float(x.get("combined_score", 0.0)), reverse=True)
-    top = ranked[:25]
-
-    return {
-        "project_buckets": buckets,
-        "top_candidates": top,
-        "ranked": ranked,
-    }
-
-
-def _write_csv(path: Path, fieldnames: List[str], rows: List[Dict[str, Any]]) -> None:
+def _write_csv(path: Path, rows: List[Dict[str, Any]], fieldnames: List[str]) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
-    with open(path, "w", newline="", encoding="utf-8") as f:
-        w = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    with tmp.open("w", newline="", encoding="utf-8") as f:
+        w = csv.DictWriter(f, fieldnames=fieldnames)
         w.writeheader()
         for r in rows:
-            w.writerow(r)
+            w.writerow({k: r.get(k) for k in fieldnames})
+    tmp.replace(path)
+
+
+def _zip_dir(src_dir: Path, zip_path: Path) -> None:
+    if zip_path.exists():
+        zip_path.unlink()
+    import zipfile
+    with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as z:
+        for p in sorted(src_dir.rglob("*")):
+            if p.is_file():
+                z.write(p, arcname=str(p.relative_to(src_dir)))
 
 
-# =========================
-# Main Pipeline (NEW CONTRACT)
-# =========================
 def run_pipeline(
     input_files: List[str],
     config: Dict[str, Any],
-    output_dir: Optional[str] = None,
-    job_id: Optional[str] = None,
+    *,
+    output_root: Optional[str] = None,
+    job_id: str,
 ) -> Dict[str, Any]:
     """
-    New stable contract:
-      - output_dir: persistent root (do not delete; holds global manifest)
-      - job_id: required by UI/API; results written under output_dir/jobs/{job_id}/...
-
-    Returns:
-      {
-        "job_id": str,
-        "job_dir": str,
-        "zip_path": Optional[str],
-        "counts": {"total": int, "success": int, "skipped": int, "failed": int},
-        "artifacts": {...}
-      }
+    Stable contract:
+      - output_root contains persistent state (manifest)
+      - job_id creates isolated job folder under output_root/jobs/{job_id}
+      - returns job_dir + zip_path + counts
     """
-    if not job_id:
-        raise ValueError("job_id is required (must be provided by UI/API).")
-
-    output_root = Path(output_dir or "/tmp/resume_eval_out").resolve()
-    output_root.mkdir(parents=True, exist_ok=True)
+    output_root_path = Path(output_root or "/tmp/resume_eval_root").resolve()
+    output_root_path.mkdir(parents=True, exist_ok=True)
 
-    # Global manifest lives outside jobs so it persists across runs
-    global_reports = output_root / GLOBAL_REPORTS_DIRNAME
-    global_reports.mkdir(parents=True, exist_ok=True)
-    manifest_path = global_reports / GLOBAL_MANIFEST_NAME
-    manifest = _read_json(manifest_path, default={"schema_version": SCHEMA_VERSION, "by_sha": {}})
+    # Persistent manifest across runs (dedupe state)
+    manifest_path = output_root_path / "processed_manifest.json"
+    manifest: Dict[str, Any] = _read_json(manifest_path, default={})
     if not isinstance(manifest, dict):
-        manifest = {"schema_version": SCHEMA_VERSION, "by_sha": {}}
-    manifest.setdefault("by_sha", {})
-
-    # Job directory
-    job_dir = (output_root / JOBS_DIRNAME / job_id).resolve()
-    if job_dir.exists():
-        # If job_id collides, fail fast. Don't silently overwrite.
-        raise RuntimeError(f"job_dir already exists for job_id={job_id}: {job_dir}")
-    job_dir.mkdir(parents=True, exist_ok=False)
-
-    # Job subfolders
-    input_dir = job_dir / INPUT_DIRNAME
-    text_dir = job_dir / TEXT_DIRNAME
-    eval_dir = job_dir / EVAL_DIRNAME
-    reports_dir = job_dir / REPORTS_DIRNAME
-
-    input_dir.mkdir(parents=True, exist_ok=True)
-    text_dir.mkdir(parents=True, exist_ok=True)
-    eval_dir.mkdir(parents=True, exist_ok=True)
-    reports_dir.mkdir(parents=True, exist_ok=True)
-
-    # Config knobs
+        manifest = {}
+
+    # Job layout
+    jobs_root = output_root_path / "jobs"
+    job_dir = jobs_root / job_id
+    input_dir = job_dir / "input"
+    text_dir = job_dir / "extracted_text"
+    eval_dir = job_dir / "evaluations"
+    reports_dir = job_dir / "reports"
+
+    for d in [input_dir, text_dir, eval_dir, reports_dir]:
+        d.mkdir(parents=True, exist_ok=True)
+
     rewrite = bool(config.get("rewrite", False))
     projects = config.get("projects") or [{"name": "STANDARD"}]
-    project_name = (projects[0] or {}).get("name", "STANDARD")
+    default_project_name = (projects[0] or {}).get("name", "STANDARD")
 
     ocr_max_pages = int(config.get("ocr_max_pages", 8))
     ocr_dpi = int(config.get("ocr_dpi", 200))
 
-    job_index: List[Dict[str, Any]] = []
+    thresholds = config.get("bucket_thresholds") or {"top": 8.0, "strong": 6.5, "maybe": 5.0}
+    top_n = int(config.get("top_n", 25))
 
-    counts = {"total": 0, "success": 0, "skipped": 0, "failed": 0}
+    per_job_index: List[Dict[str, Any]] = []
+    evaluations: List[Dict[str, Any]] = []
 
-    # Job metadata begins
-    job_json = {
-        "schema_version": SCHEMA_VERSION,
-        "job_id": job_id,
-        "created_at": _now_ts(),
-        "status": "running",
-        "config": {
-            "model": config.get("model") or os.getenv("OPENAI_MODEL") or DEFAULT_MODEL,
-            "rewrite": rewrite,
-            "projects": projects,
-            "ocr_max_pages": ocr_max_pages,
-            "ocr_dpi": ocr_dpi,
-        },
-        "counts": dict(counts),
-    }
-    _atomic_write_json(job_dir / JOB_JSON_NAME, job_json)
+    counts = {"total": 0, "success": 0, "skipped": 0, "failed": 0}
 
-    # Process each pdf
-    for pdf_path in input_files or []:
-        pdf_path = str(Path(pdf_path).resolve())
+    for src_path in input_files or []:
         counts["total"] += 1
 
-        sha = _sha256_file(pdf_path)
-        rec = _make_record_base(pdf_path, config, project_name, sha)
+        src_path = str(Path(src_path).resolve())
+        filename = os.path.basename(src_path)
 
-        # Optional: copy original into job input/
+        # Copy into job input/ (this is important for later SFTP job contract)
+        dst_pdf = input_dir / filename
         try:
-            dst_pdf = input_dir / f"{_safe_slug(Path(pdf_path).stem)}__{sha[:12]}.pdf"
-            shutil.copy2(pdf_path, dst_pdf)
-            rec["input_pdf"] = str(dst_pdf.relative_to(job_dir))
+            shutil.copy2(src_path, dst_pdf)
         except Exception:
-            rec["input_pdf"] = None
-
-        # Dedupe via global manifest
-        if (sha in manifest["by_sha"]) and (not rewrite):
-            rec["status"] = "skipped"
-            rec["error"] = "duplicate_pdf_sha256"
+            # if copy fails, still try reading original
+            dst_pdf = Path(src_path)
+
+        sha = _sha256_file(str(dst_pdf))
+
+        record = {
+            "schema_version": SCHEMA_VERSION,
+            "job_id": job_id,
+            "pdf_sha256": sha,
+            "filename": filename,
+            "candidate_name": None,
+            "project": default_project_name,
+            "model": config.get("model") or os.getenv("OPENAI_MODEL") or DEFAULT_MODEL,
+            "status": None,          # success|skipped|failed
+            "error": None,
+            "created_at": _now_ts(),
+            "output_json": None,     # relative to job_dir
+            "extracted_text": None,  # relative to job_dir
+        }
+
+        # Dedupe via persistent manifest
+        if not rewrite and sha in manifest:
+            record["status"] = "skipped"
+            record["error"] = "duplicate_pdf_sha256"
             counts["skipped"] += 1
-
-            job_index.append(rec)
-            _atomic_write_json(job_dir / JOB_INDEX_NAME, job_index)
+            per_job_index.append(record)
             continue
 
         try:
             text = extract_text_from_pdf(
-                pdf_path,
+                str(dst_pdf),
                 ocr_if_empty=True,
                 max_pages=ocr_max_pages,
                 ocr_dpi=ocr_dpi,
@@ -455,152 +373,173 @@ def run_pipeline(
             if not text.strip():
                 raise RuntimeError("No extractable text (even after OCR).")
 
-            # Write extracted text
-            text_name = f"{_safe_slug(Path(pdf_path).stem)}__{sha[:12]}.txt"
+            text_name = f"{_safe_slug(Path(filename).stem)}__{sha[:12]}.txt"
             tpath = text_dir / text_name
-            tpath.write_text(text, encoding="utf-8")
-            rec["extracted_text"] = str(tpath.relative_to(job_dir))
+            _atomic_write_text(tpath, text)
+            record["extracted_text"] = str(tpath.relative_to(job_dir))
 
-            # LLM eval
             raw = llm_evaluate(text, config)
-            ev = normalize_eval(raw, config)
+            ev = normalize_eval(raw, config, job_id=job_id, pdf_sha256=sha, filename=filename)
 
-            ev["filename"] = os.path.basename(pdf_path)
-            ev["pdf_sha256"] = sha
-            ev["job_id"] = job_id
-            ev["project"] = project_name
-
-            safe_name = _safe_slug(ev.get("candidate_name") or Path(pdf_path).stem)
+            safe_name = _safe_slug(ev.get("candidate_name") or Path(filename).stem)
             out_path = eval_dir / f"{safe_name}__{sha[:12]}.json"
-            out_path.write_text(json.dumps(ev, ensure_ascii=False, indent=2), encoding="utf-8")
-
-            rec["status"] = "success"
-            rec["candidate_name"] = ev.get("candidate_name")
-            rec["output_json"] = str(out_path.relative_to(job_dir))
+            _atomic_write_json(out_path, ev)
 
-            # Pull combined score into index row for report sorting
-            combined = (ev.get("scores") or {}).get("combined", 0.0)
-            rec["combined_score"] = float(_coerce_score(combined))
+            record["status"] = "success"
+            record["candidate_name"] = ev.get("candidate_name")
+            record["output_json"] = str(out_path.relative_to(job_dir))
 
             counts["success"] += 1
+            evaluations.append(ev)
 
-            # Update global manifest only on success
-            manifest["by_sha"][sha] = {
+            # update global manifest
+            manifest[sha] = {
                 "pdf_sha256": sha,
-                "first_seen_at": manifest["by_sha"].get(sha, {}).get("first_seen_at") or _now_ts(),
-                "last_processed_at": _now_ts(),
+                "first_seen_at": manifest.get(sha, {}).get("first_seen_at", _now_ts()),
+                "last_seen_at": _now_ts(),
                 "last_job_id": job_id,
+                "filename": filename,
                 "status": "success",
             }
-            _atomic_write_json(manifest_path, manifest)
 
         except Exception as e:
-            rec["status"] = "failed"
-            rec["error"] = f"{type(e).__name__}: {e}"
+            record["status"] = "failed"
+            record["error"] = f"{type(e).__name__}: {e}"
             counts["failed"] += 1
 
-        job_index.append(rec)
-        _atomic_write_json(job_dir / JOB_INDEX_NAME, job_index)
-
-        # update job.json counts continuously
-        job_json["counts"] = dict(counts)
-        _atomic_write_json(job_dir / JOB_JSON_NAME, job_json)
-
-    # Derived reports
-    reports = _compute_reports(job_index)
-
-    # project_buckets.json
-    project_buckets_json = reports_dir / "project_buckets.json"
-    _atomic_write_json(project_buckets_json, reports["project_buckets"])
-
-    # top_candidates.json
-    top_candidates_json = reports_dir / "top_candidates.json"
-    _atomic_write_json(top_candidates_json, reports["top_candidates"])
-
-    # candidate_analysis.csv (flat)
-    candidate_analysis_csv = reports_dir / "candidate_analysis.csv"
-    flat_rows: List[Dict[str, Any]] = []
-    for r in reports["ranked"]:
-        flat_rows.append({
-            "pdf_sha256": r.get("pdf_sha256"),
-            "filename": r.get("filename"),
-            "candidate_name": r.get("candidate_name"),
-            "project": r.get("project"),
-            "status": r.get("status"),
-            "combined_score": r.get("combined_score"),
-            "output_json": r.get("output_json"),
-            "extracted_text": r.get("extracted_text"),
-            "created_at": r.get("created_at"),
+            manifest[sha] = {
+                "pdf_sha256": sha,
+                "first_seen_at": manifest.get(sha, {}).get("first_seen_at", _now_ts()),
+                "last_seen_at": _now_ts(),
+                "last_job_id": job_id,
+                "filename": filename,
+                "status": "failed",
+                "error": record["error"],
+            }
+
+        per_job_index.append(record)
+
+    # Write per-job index
+    _atomic_write_json(job_dir / "resumes_index.json", per_job_index)
+
+    # Reports: project buckets + top candidates + candidate analysis
+    bucket_rows: List[Dict[str, Any]] = []
+    top_rows: List[Dict[str, Any]] = []
+    analysis_rows: List[Dict[str, Any]] = []
+
+    for ev in evaluations:
+        combined = float(ev.get("scores", {}).get("combined", 0.0))
+        b = _bucket_label(combined, thresholds)
+        project = (ev.get("best_project") or {}).get("project_name") or default_project_name
+
+        bucket_rows.append({
+            "job_id": job_id,
+            "pdf_sha256": ev.get("pdf_sha256"),
+            "candidate_name": ev.get("candidate_name"),
+            "seniority": ev.get("seniority"),
+            "project": project,
+            "bucket": b,
+            "combined": combined,
         })
+
+        analysis_rows.append({
+            "job_id": job_id,
+            "pdf_sha256": ev.get("pdf_sha256"),
+            "candidate_name": ev.get("candidate_name"),
+            "seniority": ev.get("seniority"),
+            "project": project,
+            "skill": ev.get("scores", {}).get("skill"),
+            "experience": ev.get("scores", {}).get("experience"),
+            "growth": ev.get("scores", {}).get("growth"),
+            "context_fit": ev.get("scores", {}).get("context_fit"),
+            "combined": combined,
+            "tags": ",".join(ev.get("tags") or []),
+        })
+
+    # sort for top candidates
+    evaluations_sorted = sorted(
+        evaluations,
+        key=lambda x: float((x.get("scores") or {}).get("combined", 0.0)),
+        reverse=True,
+    )[:max(0, top_n)]
+
+    for ev in evaluations_sorted:
+        combined = float(ev.get("scores", {}).get("combined", 0.0))
+        project = (ev.get("best_project") or {}).get("project_name") or default_project_name
+        top_rows.append({
+            "job_id": job_id,
+            "pdf_sha256": ev.get("pdf_sha256"),
+            "candidate_name": ev.get("candidate_name"),
+            "seniority": ev.get("seniority"),
+            "project": project,
+            "combined": combined,
+        })
+
+    # Write report files
+    _atomic_write_json(reports_dir / "project_buckets.json", bucket_rows)
     _write_csv(
-        candidate_analysis_csv,
-        fieldnames=list(flat_rows[0].keys()) if flat_rows else [
-            "pdf_sha256","filename","candidate_name","project","status","combined_score",
-            "output_json","extracted_text","created_at"
-        ],
-        rows=flat_rows,
+        reports_dir / "project_buckets.csv",
+        bucket_rows,
+        ["job_id", "pdf_sha256", "candidate_name", "seniority", "project", "bucket", "combined"],
     )
 
-    # project_buckets.csv (summary counts)
-    project_buckets_csv = reports_dir / "project_buckets.csv"
-    bucket_rows: List[Dict[str, Any]] = []
-    for proj, bmap in (reports["project_buckets"] or {}).items():
-        for bucket_name, items in (bmap or {}).items():
-            bucket_rows.append({
-                "project": proj,
-                "bucket": bucket_name,
-                "count": len(items),
-            })
-    _write_csv(project_buckets_csv, fieldnames=["project","bucket","count"], rows=bucket_rows)
-
-    # top_candidates.csv
-    top_candidates_csv = reports_dir / "top_candidates.csv"
-    tc_rows: List[Dict[str, Any]] = []
-    for r in reports["top_candidates"]:
-        tc_rows.append({
-            "candidate_name": r.get("candidate_name"),
-            "filename": r.get("filename"),
-            "project": r.get("project"),
-            "combined_score": r.get("combined_score"),
-            "output_json": r.get("output_json"),
-        })
+    _atomic_write_json(reports_dir / "top_candidates.json", top_rows)
     _write_csv(
-        top_candidates_csv,
-        fieldnames=["candidate_name","filename","project","combined_score","output_json"],
-        rows=tc_rows,
+        reports_dir / "top_candidates.csv",
+        top_rows,
+        ["job_id", "pdf_sha256", "candidate_name", "seniority", "project", "combined"],
     )
 
-    # artifacts.json
-    artifacts = {
+    _write_csv(
+        reports_dir / "candidate_analysis.csv",
+        analysis_rows,
+        ["job_id", "pdf_sha256", "candidate_name", "seniority", "project",
+         "skill", "experience", "growth", "context_fit", "combined", "tags"],
+    )
+
+    # Job + artifacts descriptors
+    job_json = {
         "schema_version": SCHEMA_VERSION,
         "job_id": job_id,
         "created_at": _now_ts(),
+        "model": config.get("model") or os.getenv("OPENAI_MODEL") or DEFAULT_MODEL,
+        "counts": counts,
         "paths": {
-            "job_json": JOB_JSON_NAME,
-            "resumes_index": JOB_INDEX_NAME,
-            "evaluations_dir": EVAL_DIRNAME,
-            "extracted_text_dir": TEXT_DIRNAME,
-            "reports_dir": REPORTS_DIRNAME,
-            "project_buckets_json": str(project_buckets_json.relative_to(job_dir)),
-            "project_buckets_csv": str(project_buckets_csv.relative_to(job_dir)),
-            "top_candidates_json": str(top_candidates_json.relative_to(job_dir)),
-            "top_candidates_csv": str(top_candidates_csv.relative_to(job_dir)),
-            "candidate_analysis_csv": str(candidate_analysis_csv.relative_to(job_dir)),
+            "input_dir": "input/",
+            "extracted_text_dir": "extracted_text/",
+            "evaluations_dir": "evaluations/",
+            "reports_dir": "reports/",
+        },
+    }
+    _atomic_write_json(job_dir / "job.json", job_json)
+
+    artifacts = {
+        "schema_version": SCHEMA_VERSION,
+        "job_id": job_id,
+        "files": {
+            "job_json": "job.json",
+            "resumes_index": "resumes_index.json",
+            "reports": {
+                "project_buckets_json": "reports/project_buckets.json",
+                "project_buckets_csv": "reports/project_buckets.csv",
+                "top_candidates_json": "reports/top_candidates.json",
+                "top_candidates_csv": "reports/top_candidates.csv",
+                "candidate_analysis_csv": "reports/candidate_analysis.csv",
+            },
         },
-        "counts": dict(counts),
     }
-    _atomic_write_json(job_dir / ARTIFACTS_JSON_NAME, artifacts)
+    _atomic_write_json(job_dir / "artifacts.json", artifacts)
+
+    # Persist manifest last (atomic)
+    _atomic_write_json(manifest_path, manifest)
 
-    # finalize job.json
-    job_json["status"] = "done"
-    job_json["finished_at"] = _now_ts()
-    job_json["counts"] = dict(counts)
-    _atomic_write_json(job_dir / JOB_JSON_NAME, job_json)
+    # Zip the job folder ONLY
+    zip_path = output_root_path / f"{job_id}.zip"
+    _zip_dir(job_dir, zip_path)
 
     return {
         "job_id": job_id,
         "job_dir": str(job_dir),
-        "zip_path": None,  # UI zips job_dir; worker/API might zip here later
-        "counts": dict(counts),
-        "artifacts": artifacts,
+        "zip_path": str(zip_path),
+        "counts": counts,
     }