claude-resume-kit/resume_builder/helpers/application_records.py

"""Authoritative application state and generated, read-only projections.

Run with --check to detect stale views, or --render after editing the registry.
Outcome events describe what happened; current outcome never erases an interview.
"""
from __future__ import annotations

import argparse
from contextlib import contextmanager
from datetime import date
import json
import math
import os
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
REGISTRY = ROOT / "job_scout/state/applications.json"
VIEW = ROOT / "resume_builder/reference/current_applications.md"
COHORT = ROOT / "job_scout/state/application_cohort.json"
OUTCOMES = {"open", "applied", "paused", "not_proceeding", "role_closed", "closed_no_response", "withdrawn_by_candidate", "rejected_no_interview", "rejected_after_interview", "interviewing", "offer", "accepted"}


def valid_date(value: str) -> str:
    if not isinstance(value, str) or date.fromisoformat(value).isoformat() != value:
        raise ValueError("expected ISO date YYYY-MM-DD")
    return value


def validate(data: dict) -> None:
    ids = set()
    for record in data["applications"]:
        identifier = record.get("id")
        if not identifier or identifier in ids:
            raise ValueError("application IDs must be nonempty and unique")
        ids.add(identifier)
        if record.get("outcome") not in OUTCOMES:
            raise ValueError(f"{identifier}: invalid outcome")
        for key in ("application_date", "rejection_date", "withdrawn"):
            if record.get(key):
                valid_date(record[key])
        for event in record.get("events", []):
            if event.get("date"):
                valid_date(event["date"])
        score = record.get("evidence_fit")
        if score is not None and (isinstance(score, bool) or not isinstance(score, (float, int)) or not math.isfinite(score) or not 0 <= score <= 100):
            raise ValueError(f"{identifier}: evidence_fit must be finite and between 0 and 100")
        if record.get("hard_gate") == "fail" and record.get("fit_class") not in {"stretch", "no-go"}:
            raise ValueError(f"{identifier}: failed gate requires Stretch or No-go")
        if score is not None and record.get("hard_gate") == "pass" and record.get("fit_class") in {"core", "adjacent"}:
            expected = "core" if score >= 75 else "adjacent" if score >= 60 else "stretch"
            if record["fit_class"] != expected and not record.get("fit_override_reason"):
                raise ValueError(f"{identifier}: classification differs from score; record fit_override_reason")


@contextmanager
def transaction():
    """Serialize local writers; fail clearly rather than overwrite another update."""
    lock = REGISTRY.with_suffix(".lock")
    try:
        descriptor = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    except FileExistsError as exc:
        raise ValueError(f"Application registry is locked: {lock}. Check for an active writer before removing a stale lock.") from exc
    try:
        with os.fdopen(descriptor, "w") as stream:
            stream.write(str(os.getpid()))
        data = load()
        yield data
        validate(data)
        atomic_write(REGISTRY, json.dumps(data, indent=2, ensure_ascii=False) + "\n")
        render(data)
    finally:
        lock.unlink(missing_ok=True)


def atomic_write(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary = tempfile.mkstemp(dir=path.parent, prefix=path.name + ".", suffix=".tmp")
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream:
            stream.write(text)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
    finally:
        Path(temporary).unlink(missing_ok=True)


def load() -> dict:
    return json.loads(REGISTRY.read_text(encoding="utf-8"))


def interview_reached(record: dict) -> bool:
    return bool(record.get("interview")) or any(
        event.get("type") in {"interview_invited", "interview_held"}
        for event in record.get("events", [])
    )


def projections(data: dict) -> dict[Path, str]:
    lines = ["# Current applications", "", "Generated from `job_scout/state/applications.json`; do not edit this view.",
             "Use current outcome and next action below. Imported legacy notes are historical, not current instructions.", "",
             "| Application | Current outcome | Interview reached | Next action |", "|---|---|---|---|"]
    for record in data["applications"]:
        cells = [record["label"], record["outcome"], "yes" if interview_reached(record) else "not recorded", record["next_action"]]
        lines.append("| " + " | ".join(str(cell).replace("|", " / ").replace("\n", " ") for cell in cells) + " |")
    lines.extend(["", "## Current targeting notes", "", *["- " + note for note in data.get("targeting_notes", [])], ""])
    cohort = dict(data["cohort_metadata"])
    cohort["generated_from"] = "job_scout/state/applications.json"
    cohort["applications"] = [record for record in data["applications"] if record.get("cohort_id") == cohort["cohort_id"]]
    return {VIEW: "\n".join(lines), COHORT: json.dumps(cohort, indent=2, ensure_ascii=False) + "\n"}


def render(data: dict, check: bool = False) -> list[str]:
    validate(data)
    stale = []
    for path, expected in projections(data).items():
        if not path.exists() or path.read_text(encoding="utf-8") != expected:
            stale.append(str(path.relative_to(ROOT)))
            if not check:
                atomic_write(path, expected)
    return stale


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    mode = parser.add_mutually_exclusive_group(required=True)
    mode.add_argument("--check", action="store_true")
    mode.add_argument("--render", action="store_true")
    args = parser.parse_args()
    stale = render(load(), check=args.check)
    if args.check and stale:
        print("STALE: " + ", ".join(stale))
        return 1
    print("Application views are current.")
    return 0


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