"""Evidence-grounded, privacy-minimising resume generation pipeline."""

from __future__ import annotations

import re
from collections.abc import Iterable, Mapping, Sequence
from enum import StrEnum
from pathlib import Path
from typing import Any, Literal, TypeVar, get_args

from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator

from .backend import LLMBackend
from .models import (
    CandidateProfile,
    ContentPlan,
    EvidenceItem,
    EvidenceMap,
    EvidenceMatchType,
    GenerationConfig,
    JobAnalysis,
    JobPosting,
    QualityCategory,
    QualityFinding,
    QualityReport,
    QualitySeverity,
    ResumeDraft,
    ResumeMode,
)
from .output_constraints import as_quality_findings, validate_output_constraints
from .quality import (
    RUBRIC_WEIGHTS,
    apply_deterministic_score_caps,
    compute_coverage,
    compute_evaluation_fingerprint,
    compute_evaluation_policy_fingerprint,
    compute_weighted_overall,
)
from .validators import (
    contains_blocking_posting_field,
    contains_public_blind_origin,
    validate_resume_draft,
)


ModelT = TypeVar("ModelT", bound=BaseModel)

_STAGE_PROMPTS = {
    "analyze-job": "analyze-job.md",
    "map-evidence": "map-evidence.md",
    "plan-content": "plan-content.md",
    "draft-resume": "draft-resume.md",
    "evaluate-resume": "evaluate-resume.md",
    "repair-resume": "repair-resume.md",
}
_BASE_PROMPT = "base-system.md"
_MAX_PROMPT_BYTES = 256_000
_REDACTION = "[REDACTED]"
_PRIVATE_FACT_TOKEN_MIN_LENGTH = 8
_PUBLIC_BLIND_SCHOOL_PATTERNS = (
    re.compile(
        r"(?<![가-힣A-Za-z0-9])"
        r"[가-힣A-Za-z0-9·]{2,}(?:대학교|고등학교|중학교)"
        r"(?=에서|의|를|은|는|이|가|졸업|재학|수료|[\s,.)]|$)",
        re.IGNORECASE,
    ),
    re.compile(
        r"(?<![가-힣A-Za-z0-9])"
        r"(?!최대|상대|절대|확대|세대|일대|휴대|군대|근대|현대)"
        r"[가-힣]{1,8}대"
        r"(?=에서|의|를|은|는|이|가|출신|졸업|재학|수료|전공|[\s,.)]|$)"
    ),
    re.compile(
        r"(?<![A-Za-z0-9])(?:SNU|KAIST|POSTECH|UNIST|GIST|DGIST)"
        r"(?=에서|의|를|은|는|이|가|출신|졸업|재학|수료|전공|[\s,.)]|$)",
        re.IGNORECASE,
    ),
)

# These correspond to the four release-critical areas in docs/quality-rubric.md.
MAJOR_QUALITY_CATEGORIES = frozenset(
    {
        QualityCategory.EVIDENCE,
        QualityCategory.JOB_ALIGNMENT,
        QualityCategory.KOREAN_LANGUAGE,
        QualityCategory.PRIVACY,
    }
)
MINIMUM_OVERALL_SCORE = 90.0
MINIMUM_EVIDENCE_COVERAGE = 1.0
MINIMUM_REQUIREMENT_COVERAGE = 0.80
MINIMUM_MAJOR_CATEGORY_SCORE = 80.0


class PipelineError(RuntimeError):
    """Base error for prompt, backend, or cross-stage contract failures."""


class PromptLoadError(PipelineError):
    """A required, repository-owned prompt could not be loaded safely."""


class BackendCallError(PipelineError):
    """A backend call failed or returned data outside its output schema."""


class StageIntegrityError(PipelineError):
    """A valid stage model contains broken cross-stage references."""


class PipelineStatus(StrEnum):
    PASSED = "passed"
    NEEDS_USER_INPUT = "needs_user_input"


class PipelineResult(BaseModel):
    """Approved canonical draft, or a bounded request for missing evidence."""

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    status: PipelineStatus
    analysis: JobAnalysis
    evidence_map: EvidenceMap | None = None
    content_plan: ContentPlan | None = None
    draft: ResumeDraft | None = None
    quality_report: QualityReport | None = None
    deterministic_findings: list[QualityFinding] = Field(default_factory=list)
    repair_attempts: int = Field(ge=0, le=2)
    gate_failures: list[str] = Field(default_factory=list)
    questions: list[str] = Field(default_factory=list, max_length=3)

    @model_validator(mode="after")
    def validate_outcome(self) -> "PipelineResult":
        if self.status is PipelineStatus.PASSED:
            if any(
                artifact is None
                for artifact in (
                    self.evidence_map,
                    self.content_plan,
                    self.draft,
                    self.quality_report,
                )
            ):
                raise ValueError("a passed pipeline result requires every artifact")
            if self.gate_failures:
                raise ValueError("a passed pipeline result cannot have gate failures")
            if self.questions:
                raise ValueError("a passed pipeline result cannot ask follow-up questions")
        elif not self.questions:
            raise ValueError("needs_user_input must include at least one question")
        return self

    @property
    def passed(self) -> bool:
        return self.status is PipelineStatus.PASSED


class _PromptLoader:
    """Small loader intentionally independent from the parallel prompts module."""

    def __init__(self, prompt_dir: Path) -> None:
        self._root = prompt_dir.expanduser().resolve()
        if not self._root.is_dir():
            raise PromptLoadError(f"prompt directory does not exist: {self._root}")

    def read(self, filename: str) -> str:
        if Path(filename).name != filename:
            raise PromptLoadError("prompt filename must not contain a path")
        path = (self._root / filename).resolve()
        try:
            path.relative_to(self._root)
        except ValueError as exc:
            raise PromptLoadError("prompt path escapes prompt directory") from exc
        try:
            size = path.stat().st_size
            if size > _MAX_PROMPT_BYTES:
                raise PromptLoadError(f"prompt is unexpectedly large: {filename}")
            content = path.read_text(encoding="utf-8").strip()
        except (OSError, UnicodeError) as exc:
            raise PromptLoadError(f"could not read required prompt: {filename}") from exc
        if not content:
            raise PromptLoadError(f"required prompt is empty: {filename}")
        return content


class _PrivacyContext:
    def __init__(
        self,
        *,
        exact_tokens: Iterable[str],
        phone_numbers: Iterable[str],
    ) -> None:
        # Longest first prevents a short token from partially masking a longer one.
        self.exact_tokens = tuple(
            sorted(
                {token.strip() for token in exact_tokens if token and token.strip()},
                key=len,
                reverse=True,
            )
        )
        self.exact_patterns = tuple(
            _private_token_pattern(token) for token in self.exact_tokens
        )
        self.phone_patterns = tuple(
            re.compile(r"(?<!\d)" + r"[\s().-]*".join(map(re.escape, digits)) + r"(?!\d)")
            for value in phone_numbers
            if len(digits := re.sub(r"\D", "", value)) >= 8
        )

    def redact(self, value: Any) -> Any:
        if isinstance(value, str):
            redacted = value
            for pattern in self.exact_patterns:
                redacted = pattern.sub(_REDACTION, redacted)
            for pattern in self.phone_patterns:
                redacted = pattern.sub(_REDACTION, redacted)
            return redacted
        if isinstance(value, Mapping):
            return {str(key): self.redact(item) for key, item in value.items()}
        if isinstance(value, tuple):
            return [self.redact(item) for item in value]
        if isinstance(value, list):
            return [self.redact(item) for item in value]
        if isinstance(value, set):
            return [self.redact(item) for item in sorted(value, key=str)]
        return value


def _private_token_pattern(token: str) -> re.Pattern[str]:
    r"""Compile a PII token without missing Korean postpositions.

    Python's ``\w`` includes Hangul, so a generic word boundary does not match
    ``김하늘은``.  Multi-character Hangul identity tokens are therefore matched
    as literal substrings.  Single-character names retain boundaries to avoid
    erasing common Korean syllables throughout an otherwise safe payload.
    """

    if re.search(r"[가-힣]", token) and len(token) >= 2:
        compact_token = re.sub(r"\s+", "", token)
        flexible_token = r"\s*".join(
            re.escape(character) for character in compact_token
        )
        return re.compile(
            r"(?<![가-힣A-Za-z0-9])"
            + flexible_token
            + r"(?=(?:은|는|이|가|을|를|의|에게|께서|으로|입니다|이라고|"
            r"[\s,.)]|$))",
            flags=re.I,
        )
    prefix = r"(?<!\w)" if token[0].isalnum() else ""
    suffix = r"(?!\w)" if token[-1].isalnum() else ""
    return re.compile(prefix + re.escape(token) + suffix, flags=re.I)


def _model_payload(model: BaseModel) -> dict[str, Any]:
    # ``exclude_computed_fields`` is newer than the project's Pydantic floor.
    # Excluding declared computed names explicitly keeps compatibility with
    # Pydantic 2.10 while avoiding read-only values in LLM payloads.
    dumped = model.model_dump(mode="json")
    return _strip_computed_fields(dumped, type(model))


def _nested_model_types(annotation: Any) -> tuple[type[BaseModel], ...]:
    if isinstance(annotation, type) and issubclass(annotation, BaseModel):
        return (annotation,)
    discovered: list[type[BaseModel]] = []
    for argument in get_args(annotation):
        for model_type in _nested_model_types(argument):
            if model_type not in discovered:
                discovered.append(model_type)
    return tuple(discovered)


def _strip_computed_fields(
    value: Mapping[str, Any], model_type: type[BaseModel]
) -> dict[str, Any]:
    """Remove only declared read-only fields, preserving strict extra checks."""

    cleaned = dict(value)
    for field_name in model_type.model_computed_fields:
        cleaned.pop(field_name, None)
    for field_name, field in model_type.model_fields.items():
        if field_name not in cleaned:
            continue
        nested_types = _nested_model_types(field.annotation)
        if not nested_types:
            continue
        nested_value = cleaned[field_name]
        if isinstance(nested_value, Mapping):
            for nested_type in nested_types:
                nested_value = _strip_computed_fields(nested_value, nested_type)
            cleaned[field_name] = nested_value
        elif isinstance(nested_value, (list, tuple)):
            items: list[Any] = []
            for item in nested_value:
                if isinstance(item, Mapping):
                    for nested_type in nested_types:
                        item = _strip_computed_fields(item, nested_type)
                items.append(item)
            cleaned[field_name] = items
    return cleaned


def _visible_facts(
    profile: CandidateProfile,
    config: GenerationConfig,
    analysis: JobAnalysis | None = None,
) -> tuple[EvidenceItem, ...]:
    """Return only evidence that may cross the LLM boundary."""

    # Consent controls whether a renderer may insert a sensitive value; it does
    # not grant an external generation model access to that value.  Keeping this
    # boundary unconditional also prevents an employer-form toggle from
    # silently broadening the model's data access.
    visible: list[EvidenceItem] = []
    for fact in profile.facts:
        if fact.confidential or fact.sensitive_category is not None:
            continue
        if config.resume_mode is ResumeMode.PUBLIC_BLIND:
            blind_text = " ".join(
                [
                    fact.content,
                    *(str(key) for key in fact.metrics),
                    *(str(value) for value in fact.metrics.values()),
                    *fact.keywords,
                ]
            )
            if any(
                pattern.search(blind_text)
                for pattern in _PUBLIC_BLIND_SCHOOL_PATTERNS
            ) or contains_public_blind_origin(blind_text):
                continue
        fact_text = " ".join(
            [
                fact.content,
                *(str(key) for key in fact.metrics),
                *(str(value) for value in fact.metrics.values()),
                *fact.keywords,
            ]
        )
        if (
            analysis is not None
            and contains_blocking_posting_field(fact_text, profile, analysis)
        ):
            continue
        visible.append(fact)
    return tuple(visible)


def _privacy_context(
    profile: CandidateProfile,
    excluded_facts: Sequence[EvidenceItem],
) -> _PrivacyContext:
    contact = profile.contact
    identity_tokens = [
        profile.name,
        profile.name_en or "",
        contact.email or "",
        *contact.links,
    ]
    # If an excluded fact is echoed by a malformed draft, it must still not be
    # sent back to the judge or repairer.  IDs are intentionally not tokens:
    # reference integrity rejects them and common short IDs can collide with job
    # text.  The actual private fact values are redacted.
    for fact in excluded_facts:
        private_values = [fact.content, fact.source_reference or ""]
        identity_tokens.extend(
            value
            for value in private_values
            if len(value.strip()) >= _PRIVATE_FACT_TOKEN_MIN_LENGTH
        )
    return _PrivacyContext(
        exact_tokens=identity_tokens,
        phone_numbers=[contact.phone] if contact.phone else [],
    )


def _candidate_facts_payload(facts: Sequence[EvidenceItem]) -> list[dict[str, Any]]:
    """Project only generation-relevant fields across the model boundary."""

    allowed_fields = (
        "evidence_id",
        "category",
        "content",
        "date_range",
        "verification_status",
        "metrics",
        "keywords",
    )
    payload: list[dict[str, Any]] = []
    for fact in facts:
        dumped = _model_payload(fact)
        payload.append({field: dumped[field] for field in allowed_fields})
    return payload


class ResumePipeline:
    """Orchestrate structured generation, validation, judging, and repair."""

    def __init__(
        self,
        backend: LLMBackend,
        *,
        prompt_dir: str | Path | None = None,
        max_repair_attempts: int = 2,
    ) -> None:
        if not 0 <= max_repair_attempts <= 2:
            raise ValueError("max_repair_attempts must be between 0 and 2")
        self._backend = backend
        # Keep runtime assets inside the import package so wheel installations do
        # not depend on a repository-level directory that is absent after install.
        default_prompt_dir = Path(__file__).resolve().with_name("prompt_templates")
        self._prompts = _PromptLoader(
            Path(prompt_dir) if prompt_dir is not None else default_prompt_dir
        )
        self._system_prompt = self._prompts.read(_BASE_PROMPT)
        self._task_prompts = {
            stage: self._prompts.read(filename)
            for stage, filename in _STAGE_PROMPTS.items()
        }
        self._evaluation_policy_fingerprint = (
            compute_evaluation_policy_fingerprint(
                system_prompt=self._system_prompt,
                evaluator_prompt=self._task_prompts["evaluate-resume"],
            )
        )
        self._max_repair_attempts = max_repair_attempts

    def run(
        self,
        profile: CandidateProfile | Mapping[str, Any],
        posting: JobPosting | Mapping[str, Any],
        config: GenerationConfig | Mapping[str, Any],
    ) -> PipelineResult:
        """Generate and gate one canonical resume draft."""

        validated_profile = CandidateProfile.model_validate(profile)
        validated_posting = JobPosting.model_validate(posting)
        validated_config = GenerationConfig.model_validate(config)
        if not validated_config.strict_evidence:
            raise PipelineError(
                "release pipeline requires strict_evidence=true; false is lint-only"
            )
        validated_config.assert_profile_compatible(validated_profile)

        initial_visible_facts = _visible_facts(
            validated_profile, validated_config
        )
        initial_visible_ids = {
            fact.evidence_id for fact in initial_visible_facts
        }
        initial_excluded_facts = tuple(
            fact
            for fact in validated_profile.facts
            if fact.evidence_id not in initial_visible_ids
        )
        privacy = _privacy_context(validated_profile, initial_excluded_facts)

        analysis = self._invoke(
            "analyze-job",
            JobAnalysis,
            {"job_posting": _model_payload(validated_posting)},
            privacy,
        )
        self._integrity(
            "analyze-job", lambda: analysis.assert_matches_posting(validated_posting)
        )

        # Posting-specific blind/redaction rules are known only after analysis.
        # Recompute the allowlist before any candidate fact crosses the model
        # boundary, then use the same rules again during final validation.
        visible_facts = _visible_facts(
            validated_profile, validated_config, analysis
        )
        visible_ids = {fact.evidence_id for fact in visible_facts}
        excluded_facts = tuple(
            fact
            for fact in validated_profile.facts
            if fact.evidence_id not in visible_ids
        )
        privacy = _privacy_context(validated_profile, excluded_facts)
        facts_payload = _candidate_facts_payload(visible_facts)

        if not visible_facts:
            return PipelineResult(
                status=PipelineStatus.NEEDS_USER_INPUT,
                analysis=analysis,
                repair_attempts=0,
                gate_failures=["no_generation_safe_evidence"],
                questions=[
                    "지원 직무와 관련된 경력·프로젝트·교육 사실을 최소 1개 제공해 주세요."
                ],
            )

        evidence_map = self._invoke(
            "map-evidence",
            EvidenceMap,
            {
                "job_analysis": _model_payload(analysis),
                "candidate_facts": facts_payload,
            },
            privacy,
        )
        self._integrity(
            "map-evidence",
            lambda: evidence_map.assert_referential_integrity(
                validated_profile, analysis
            ),
        )
        self._assert_visible_references(
            "map-evidence",
            (
                evidence_id
                for match in evidence_map.matches
                for evidence_id in match.evidence_ids
            ),
            visible_ids,
        )

        if all(
            match.match_type is EvidenceMatchType.GAP
            for match in evidence_map.matches
        ):
            required_by_id = {
                requirement.requirement_id: requirement
                for requirement in analysis.requirements
            }
            questions = [
                (
                    f"{required_by_id[match.requirement_id].text!r}을(를) 입증할 "
                    "구체적인 경험·기간·역할·결과가 있나요?"
                )
                for match in evidence_map.matches
                if match.requirement_id in required_by_id
            ][:3]
            return PipelineResult(
                status=PipelineStatus.NEEDS_USER_INPUT,
                analysis=analysis,
                evidence_map=evidence_map,
                repair_attempts=0,
                gate_failures=["all_requirements_gap"],
                questions=questions
                or ["공고 요건과 연결할 수 있는 검증 가능한 경험을 제공해 주세요."],
            )

        content_plan = self._invoke(
            "plan-content",
            ContentPlan,
            {
                "candidate_id": validated_profile.candidate_id,
                "job_analysis": _model_payload(analysis),
                "evidence_map": _model_payload(evidence_map),
                "generation_config": _model_payload(validated_config),
            },
            privacy,
        )
        self._integrity(
            "plan-content",
            lambda: content_plan.assert_referential_integrity(
                validated_profile, analysis
            ),
        )
        self._integrity(
            "plan-content",
            lambda: content_plan.assert_matches_evidence_map(evidence_map),
        )
        if content_plan.mode is not validated_config.resume_mode:
            raise StageIntegrityError(
                "plan-content: content plan mode does not match generation config"
            )
        self._assert_visible_references(
            "plan-content",
            (
                evidence_id
                for section in content_plan.sections
                for evidence_id in section.evidence_ids
            ),
            visible_ids,
        )

        draft = self._invoke(
            "draft-resume",
            ResumeDraft,
            {
                "candidate_id": validated_profile.candidate_id,
                "candidate_facts": facts_payload,
                "job_analysis": _model_payload(analysis),
                "content_plan": _model_payload(content_plan),
                "generation_config": _model_payload(validated_config),
            },
            privacy,
        )
        self._assert_draft_integrity(
            draft,
            profile=validated_profile,
            analysis=analysis,
            evidence_map=evidence_map,
            plan=content_plan,
            config=validated_config,
            visible_ids=visible_ids,
            stage="draft-resume",
        )
        repair_evidence_ceiling = {
            evidence_id
            for section in draft.sections
            for claim in section.claims
            for evidence_id in claim.evidence_ids
        }

        repairs = 0
        while True:
            deterministic_findings = self._deterministic_findings(
                validated_profile,
                draft,
                validated_config,
                analysis,
                visible_ids,
            )
            report = self._invoke(
                "evaluate-resume",
                QualityReport,
                {
                    "candidate_facts": facts_payload,
                    "job_analysis": _model_payload(analysis),
                    "resume_draft": _model_payload(draft),
                    "deterministic_findings": [
                        _model_payload(finding)
                        for finding in deterministic_findings
                    ],
                    "quality_rubric": self._quality_rubric(validated_config),
                },
                privacy,
            )
            # The backend judges content; the trusted harness binds the report
            # to the exact canonical draft and computes objective coverage after
            # structured validation.  Neither value is trusted to the judge.
            coverage = compute_coverage(
                draft, analysis, evidence_map, validated_profile
            )
            trusted_category_scores = apply_deterministic_score_caps(
                report.category_scores, deterministic_findings
            )
            try:
                weighted_overall = compute_weighted_overall(trusted_category_scores)
            except ValueError as exc:
                raise StageIntegrityError(
                    "evaluate-resume: quality report omitted weighted rubric categories"
                ) from exc
            evaluation_fingerprint = compute_evaluation_fingerprint(
                validated_profile,
                draft,
                validated_posting,
                analysis,
                evidence_map,
                content_plan,
                validated_config,
                policy_fingerprint=self._evaluation_policy_fingerprint,
            )
            report = report.model_copy(
                update={
                    "category_scores": trusted_category_scores,
                    "draft_fingerprint": draft.fingerprint(),
                    "evaluation_fingerprint": evaluation_fingerprint,
                    "overall_score": weighted_overall,
                    "evidence_coverage": coverage.evidence,
                    "requirement_coverage": coverage.requirements,
                    "minimum_score": max(
                        MINIMUM_OVERALL_SCORE,
                        validated_config.minimum_quality_score,
                    ),
                    "minimum_evidence_coverage": max(
                        MINIMUM_EVIDENCE_COVERAGE,
                        validated_config.minimum_evidence_coverage,
                    ),
                    "minimum_requirement_coverage": max(
                        MINIMUM_REQUIREMENT_COVERAGE,
                        validated_config.minimum_requirement_coverage,
                    ),
                }
            )
            self._assert_report_integrity(
                report,
                draft,
                visible_ids,
                evaluation_fingerprint=evaluation_fingerprint,
                stage="evaluate-resume",
            )
            gate_failures = self._gate_failures(
                deterministic_findings, report, validated_config
            )
            if not gate_failures:
                return PipelineResult(
                    status=PipelineStatus.PASSED,
                    analysis=analysis,
                    evidence_map=evidence_map,
                    content_plan=content_plan,
                    draft=draft,
                    quality_report=report,
                    deterministic_findings=deterministic_findings,
                    repair_attempts=repairs,
                )

            repair_findings = self._claim_repair_findings(
                deterministic_findings, report.findings, draft
            )
            if repairs >= self._max_repair_attempts or not repair_findings:
                return PipelineResult(
                    status=PipelineStatus.NEEDS_USER_INPUT,
                    analysis=analysis,
                    evidence_map=evidence_map,
                    content_plan=content_plan,
                    draft=draft,
                    quality_report=report,
                    deterministic_findings=deterministic_findings,
                    repair_attempts=repairs,
                    gate_failures=gate_failures,
                    questions=self._questions(
                        repair_findings,
                        gate_failures,
                        report,
                    ),
                )

            repaired = self._invoke(
                "repair-resume",
                ResumeDraft,
                {
                    "candidate_facts": facts_payload,
                    "resume_draft": _model_payload(draft),
                    "approved_findings": [
                        _model_payload(finding) for finding in repair_findings
                    ],
                    "generation_config": _model_payload(validated_config),
                },
                privacy,
            )
            targeted_claim_ids = {
                finding.claim_id
                for finding in repair_findings
                if finding.claim_id is not None
            }
            self._assert_draft_integrity(
                repaired,
                profile=validated_profile,
                analysis=analysis,
                evidence_map=evidence_map,
                plan=content_plan,
                config=validated_config,
                visible_ids=visible_ids,
                stage="repair-resume",
            )
            self._assert_repair_scope(
                previous=draft,
                repaired=repaired,
                targeted_claim_ids=targeted_claim_ids,
                evidence_ceiling=repair_evidence_ceiling,
            )
            draft = repaired
            repairs += 1

    def generate(
        self,
        profile: CandidateProfile | Mapping[str, Any],
        posting: JobPosting | Mapping[str, Any],
        config: GenerationConfig | Mapping[str, Any],
    ) -> PipelineResult:
        """Compatibility-friendly synonym for :meth:`run`."""

        return self.run(profile, posting, config)

    def _invoke(
        self,
        stage: str,
        output_model: type[ModelT],
        user_payload: Mapping[str, Any],
        privacy: _PrivacyContext,
    ) -> ModelT:
        safe_payload = privacy.redact(user_payload)
        try:
            raw = self._backend.complete_json(
                stage=stage,
                system_prompt=self._system_prompt,
                task_prompt=self._task_prompts[stage],
                user_payload=safe_payload,
                output_model=output_model,
            )
        except Exception as exc:
            raise BackendCallError(f"{stage}: backend call failed") from exc

        if isinstance(raw, BaseModel):
            candidate: Any = {
                field_name: getattr(raw, field_name)
                for field_name in type(raw).model_fields
            }
        elif isinstance(raw, Mapping):
            candidate = _strip_computed_fields(raw, output_model)
        else:
            raise BackendCallError(
                f"{stage}: backend returned neither a model nor a mapping"
            )
        try:
            return output_model.model_validate(candidate)
        except (ValidationError, TypeError, ValueError) as exc:
            # Do not interpolate raw output: it may contain candidate PII.
            raise BackendCallError(
                f"{stage}: backend output failed {output_model.__name__} validation"
            ) from exc

    @staticmethod
    def _integrity(stage: str, check: Any) -> None:
        try:
            check()
        except (ValidationError, TypeError, ValueError) as exc:
            raise StageIntegrityError(f"{stage}: {exc}") from exc

    @staticmethod
    def _assert_visible_references(
        stage: str,
        references: Iterable[str],
        visible_ids: set[str],
    ) -> None:
        hidden = sorted(set(references) - visible_ids)
        if hidden:
            raise StageIntegrityError(
                f"{stage}: references evidence withheld by privacy policy: {hidden}"
            )

    def _assert_draft_integrity(
        self,
        draft: ResumeDraft,
        *,
        profile: CandidateProfile,
        analysis: JobAnalysis,
        evidence_map: EvidenceMap,
        plan: ContentPlan,
        config: GenerationConfig,
        visible_ids: set[str],
        stage: str,
    ) -> None:
        self._integrity(
            stage, lambda: draft.assert_referential_integrity(profile, analysis)
        )
        self._integrity(
            stage, lambda: draft.assert_matches_evidence_map(evidence_map)
        )
        if draft.mode is not config.resume_mode:
            raise StageIntegrityError(
                f"{stage}: draft mode does not match generation config"
            )
        if draft.mode is not plan.mode:
            raise StageIntegrityError(f"{stage}: draft mode does not match content plan")

        plan_by_id = {section.section_id: section for section in plan.sections}
        draft_by_id = {section.section_id: section for section in draft.sections}
        if set(plan_by_id) != set(draft_by_id):
            raise StageIntegrityError(
                f"{stage}: draft sections do not match planned sections"
            )
        all_references: list[str] = []
        for section_id, section in draft_by_id.items():
            planned = plan_by_id[section_id]
            if section.section_type is not planned.section_type:
                raise StageIntegrityError(
                    f"{stage}: section {section_id!r} changed planned type"
                )
            if section.order != planned.order:
                raise StageIntegrityError(
                    f"{stage}: section {section_id!r} changed planned order"
                )
            if len(section.claims) > planned.bullet_budget:
                raise StageIntegrityError(
                    f"{stage}: section {section_id!r} exceeds bullet budget"
                )
            allowed_evidence = set(planned.evidence_ids)
            allowed_requirements = set(planned.requirement_ids)
            for claim in section.claims:
                all_references.extend(claim.evidence_ids)
                if not set(claim.evidence_ids) <= allowed_evidence:
                    raise StageIntegrityError(
                        f"{stage}: claim {claim.claim_id!r} uses unplanned evidence"
                    )
                if not set(claim.requirement_ids) <= allowed_requirements:
                    raise StageIntegrityError(
                        f"{stage}: claim {claim.claim_id!r} uses unplanned requirements"
                    )
        self._assert_visible_references(stage, all_references, visible_ids)

    def _deterministic_findings(
        self,
        profile: CandidateProfile,
        draft: ResumeDraft,
        config: GenerationConfig,
        analysis: JobAnalysis,
        visible_ids: set[str],
    ) -> list[QualityFinding]:
        try:
            raw_findings = validate_resume_draft(
                profile, draft, config, analysis=analysis
            )
            findings = [QualityFinding.model_validate(item) for item in raw_findings]
            findings.extend(
                as_quality_findings(
                    validate_output_constraints(
                        draft,
                        config,
                        analysis=analysis,
                    )
                )
            )
        except (ValidationError, TypeError, ValueError) as exc:
            raise StageIntegrityError(
                "deterministic-validate: validator returned invalid findings"
            ) from exc
        self._assert_findings_integrity(
            findings, draft, visible_ids, stage="deterministic-validate"
        )
        return findings

    def _assert_report_integrity(
        self,
        report: QualityReport,
        draft: ResumeDraft,
        visible_ids: set[str],
        *,
        evaluation_fingerprint: str,
        stage: str,
    ) -> None:
        if report.draft_id != draft.draft_id:
            raise StageIntegrityError(
                f"{stage}: quality report references a different draft"
            )
        if report.draft_fingerprint != draft.fingerprint():
            raise StageIntegrityError(
                f"{stage}: quality report fingerprint does not match draft content"
            )
        if report.evaluation_fingerprint != evaluation_fingerprint:
            raise StageIntegrityError(
                f"{stage}: quality report fingerprint does not match evaluation context"
            )
        self._assert_findings_integrity(report.findings, draft, visible_ids, stage)

    @staticmethod
    def _assert_findings_integrity(
        findings: Sequence[QualityFinding],
        draft: ResumeDraft,
        visible_ids: set[str],
        stage: str,
    ) -> None:
        known_claims = {
            claim.claim_id for section in draft.sections for claim in section.claims
        }
        errors: list[str] = []
        for finding in findings:
            if finding.claim_id is None and finding.location is None:
                errors.append(
                    f"finding {finding.finding_id!r} has no claim_id or location"
                )
            if finding.claim_id is not None and finding.claim_id not in known_claims:
                errors.append(
                    f"finding {finding.finding_id!r} references unknown claim"
                )
            hidden = sorted(set(finding.evidence_ids) - visible_ids)
            if hidden:
                errors.append(
                    f"finding {finding.finding_id!r} references unknown evidence {hidden}"
                )
        if errors:
            raise StageIntegrityError(f"{stage}: {'; '.join(errors)}")

    @staticmethod
    def _quality_rubric(config: GenerationConfig) -> dict[str, Any]:
        return {
            "dimension_weights": {
                category.value: weight
                for category, weight in RUBRIC_WEIGHTS.items()
            },
            "minimum_overall_score": max(
                MINIMUM_OVERALL_SCORE, config.minimum_quality_score
            ),
            "minimum_evidence_coverage": max(
                MINIMUM_EVIDENCE_COVERAGE, config.minimum_evidence_coverage
            ),
            "minimum_requirement_coverage": max(
                MINIMUM_REQUIREMENT_COVERAGE,
                config.minimum_requirement_coverage,
            ),
            "minimum_major_category_score": MINIMUM_MAJOR_CATEGORY_SCORE,
            "major_categories": sorted(
                category.value for category in MAJOR_QUALITY_CATEGORIES
            ),
        }

    @staticmethod
    def _gate_failures(
        deterministic: Sequence[QualityFinding],
        report: QualityReport,
        config: GenerationConfig,
    ) -> list[str]:
        failures: list[str] = []
        blocking_rules = sorted(
            {finding.code for finding in deterministic if finding.blocking}
        )
        if blocking_rules:
            failures.append(
                "deterministic_blocking:" + ",".join(blocking_rules)
            )
        judge_blocking = sorted(
            {finding.code for finding in report.findings if finding.blocking}
        )
        if judge_blocking:
            failures.append("judge_blocking:" + ",".join(judge_blocking))

        required_score = max(MINIMUM_OVERALL_SCORE, config.minimum_quality_score)
        if report.overall_score < required_score:
            failures.append(
                f"overall_score:{report.overall_score:g}<{required_score:g}"
            )
        required_evidence = max(
            MINIMUM_EVIDENCE_COVERAGE, config.minimum_evidence_coverage
        )
        if report.evidence_coverage < required_evidence:
            failures.append(
                "evidence_coverage:"
                f"{report.evidence_coverage:g}<{required_evidence:g}"
            )
        required_requirements = max(
            MINIMUM_REQUIREMENT_COVERAGE,
            config.minimum_requirement_coverage,
        )
        if report.requirement_coverage < required_requirements:
            failures.append(
                "requirement_coverage:"
                f"{report.requirement_coverage:g}<{required_requirements:g}"
            )
        for category in sorted(MAJOR_QUALITY_CATEGORIES, key=lambda item: item.value):
            score = report.category_scores.get(category)
            if score is None:
                failures.append(f"major_category_missing:{category.value}")
            elif score < MINIMUM_MAJOR_CATEGORY_SCORE:
                failures.append(
                    f"major_category:{category.value}:"
                    f"{score:g}<{MINIMUM_MAJOR_CATEGORY_SCORE:g}"
                )
        return failures

    @staticmethod
    def _claim_repair_findings(
        deterministic: Sequence[QualityFinding],
        judged: Sequence[QualityFinding],
        draft: ResumeDraft,
    ) -> list[QualityFinding]:
        known_claims = {
            claim.claim_id for section in draft.sections for claim in section.claims
        }
        selected: list[QualityFinding] = []
        seen: set[tuple[str, str | None, str]] = set()
        for finding in [*deterministic, *judged]:
            if (
                finding.claim_id not in known_claims
                or finding.severity is QualitySeverity.INFO
            ):
                continue
            fingerprint = (finding.code, finding.claim_id, finding.message)
            if fingerprint in seen:
                continue
            seen.add(fingerprint)
            selected.append(finding)
        return selected

    @staticmethod
    def _assert_repair_scope(
        *,
        previous: ResumeDraft,
        repaired: ResumeDraft,
        targeted_claim_ids: set[str],
        evidence_ceiling: set[str],
    ) -> None:
        if (
            previous.candidate_id != repaired.candidate_id
            or previous.posting_id != repaired.posting_id
            or previous.mode is not repaired.mode
            or previous.title != repaired.title
        ):
            raise StageIntegrityError(
                "repair-resume: claim repair changed draft-level identity or metadata"
            )
        previous_sections = {section.section_id: section for section in previous.sections}
        repaired_sections = {section.section_id: section for section in repaired.sections}
        if set(previous_sections) != set(repaired_sections):
            raise StageIntegrityError("repair-resume: claim repair changed section set")

        for section_id, old_section in previous_sections.items():
            new_section = repaired_sections[section_id]
            if (
                old_section.section_type is not new_section.section_type
                or old_section.heading != new_section.heading
                or old_section.order != new_section.order
            ):
                raise StageIntegrityError(
                    "repair-resume: claim repair changed section metadata"
                )
            old_claims = {claim.claim_id: claim for claim in old_section.claims}
            new_claims = {claim.claim_id: claim for claim in new_section.claims}
            if set(old_claims) != set(new_claims):
                raise StageIntegrityError(
                    "repair-resume: claim repair changed claim set"
                )
            for claim_id, old_claim in old_claims.items():
                new_claim = new_claims[claim_id]
                if old_claim.order != new_claim.order:
                    raise StageIntegrityError(
                        "repair-resume: claim repair changed claim order"
                    )
                if claim_id not in targeted_claim_ids and old_claim != new_claim:
                    raise StageIntegrityError(
                        "repair-resume: non-targeted claim was modified"
                    )
                if not set(new_claim.evidence_ids) <= evidence_ceiling:
                    raise StageIntegrityError(
                        "repair-resume: repair introduced evidence outside the "
                        "original draft"
                    )

    @staticmethod
    def _questions(
        repair_findings: Sequence[QualityFinding],
        gate_failures: Sequence[str],
        report: QualityReport,
    ) -> list[str]:
        questions: list[str] = []
        seen_claims: set[str] = set()
        for finding in repair_findings:
            claim_id = finding.claim_id
            if claim_id is None or claim_id in seen_claims:
                continue
            seen_claims.add(claim_id)
            questions.append(
                f"{claim_id} 주장을 보완할 수 있는 검증 가능한 사실이나 "
                "정확한 수치를 제공해 주시겠습니까?"
            )
            if len(questions) == 3:
                return questions

        if report.evidence_coverage < MINIMUM_EVIDENCE_COVERAGE:
            questions.append(
                "근거가 연결되지 않은 주장을 뒷받침하거나 삭제할 수 있도록 "
                "추가 사실을 제공해 주시겠습니까?"
            )
        if len(questions) < 3 and any(
            failure.startswith("requirement_coverage:") for failure in gate_failures
        ):
            questions.append(
                "아직 다루지 못한 직무 요건과 관련된 실제 경험이나 산출물이 "
                "있다면 제공해 주시겠습니까?"
            )
        if len(questions) < 3:
            deficient_categories = [
                category.value
                for category in sorted(
                    MAJOR_QUALITY_CATEGORIES, key=lambda item: item.value
                )
                if report.category_scores.get(category, -1)
                < MINIMUM_MAJOR_CATEGORY_SCORE
            ]
            if deficient_categories:
                questions.append(
                    "주요 품질 영역("
                    + ", ".join(deficient_categories)
                    + ")을 보완할 추가 근거를 제공해 주시겠습니까?"
                )
        if not questions:
            questions.append(
                "품질 기준을 충족하려면 어떤 주장을 유지해야 하는지와 이를 "
                "뒷받침할 추가 근거를 확인해 주시겠습니까?"
            )
        # Stable de-duplication and a hard API bound.
        return list(dict.fromkeys(questions))[:3]


def run_pipeline(
    backend: LLMBackend,
    profile: CandidateProfile | Mapping[str, Any],
    posting: JobPosting | Mapping[str, Any],
    config: GenerationConfig | Mapping[str, Any],
    *,
    prompt_dir: str | Path | None = None,
    max_repair_attempts: Literal[0, 1, 2] = 2,
) -> PipelineResult:
    """One-call convenience wrapper around :class:`ResumePipeline`."""

    return ResumePipeline(
        backend,
        prompt_dir=prompt_dir,
        max_repair_attempts=max_repair_attempts,
    ).run(profile, posting, config)


__all__ = [
    "BackendCallError",
    "LLMBackend",
    "MAJOR_QUALITY_CATEGORIES",
    "PipelineError",
    "PipelineResult",
    "PipelineStatus",
    "PromptLoadError",
    "ResumePipeline",
    "StageIntegrityError",
    "run_pipeline",
]