resume-agent/backend/app/resume_expansion.py

"""Light entry expansion: pure LLM expander, fallback composition, and factory."""

from __future__ import annotations

import logging
import time
from typing import Any

from .claim_validator import partition_entry_text, quantified_fact_contexts
from .entry_expander import EntryExpander, RuleBasedEntryExpander, normalize_bullet_description
from .fact_coverage import (
    FactRequirement,
    classify_fact_requirements,
    hard_fact_is_preserved,
    missing_hard_facts,
)
from .llm_services import (
    LLMServiceError,
    OpenAICompatibleStructuredClient,
    StrictSchema,
    log_ai_event,
)
from .resume_expansion_prompts import (
    _EDUCATION_PROMPT,
    _EXPANSION_REPAIR_PROMPT,
    _repair_prompt,
    _system_prompt,
)
from .settings import Settings

__all__ = [
    "EntryExpansionOutput",
    "OpenAIEntryExpander",
    "FallbackEntryExpander",
    "build_expander",
    "_EDUCATION_PROMPT",
    "_EXPANSION_REPAIR_PROMPT",
    "_system_prompt",
    "_entry_fact_ledger",
    "_entry_facts",
]


class EntryExpansionOutput(StrictSchema):
    optimized_description: str


class OpenAIEntryExpander:
    """LLM expander over user-confirmed facts only (no retrieval)."""

    _MIN_REPAIR_SECONDS = 6.0

    def __init__(self, completion: Any, *, timeout_seconds: float | None = None) -> None:
        self.completion = completion
        settings = getattr(completion, "settings", None)
        configured_timeout = getattr(settings, "light_entry_timeout_seconds", None)
        self.timeout_seconds = timeout_seconds or configured_timeout

    def expand(self, entry: dict[str, Any], *, context: dict[str, Any]) -> dict[str, Any]:
        facts_text = _entry_facts(entry)
        fact_ledger = _entry_fact_ledger(entry)
        hard_required_facts, _ = classify_fact_requirements(fact_ledger)
        entry_type = str(context.get("entry_type") or "")
        primary_description = str(entry.get("description") or "").strip()
        started_at = time.perf_counter()
        output: EntryExpansionOutput = self._complete(
            schema_name="entry_expansion",
            system_prompt=_system_prompt(entry_type),
            payload=self._base_payload(
                facts_text=facts_text,
                primary_description=primary_description,
                entry_type=entry_type,
                context=context,
                hard_required_facts=hard_required_facts,
            ),
            remaining_seconds=self._remaining_seconds(started_at),
        )

        candidate = _normalize_candidate(output.optimized_description, entry_type)
        repair_reason: str | None = None
        if not candidate and primary_description:
            repair_reason = "empty_result"
            remaining_seconds = self._remaining_seconds(started_at)
            if remaining_seconds is None or remaining_seconds >= self._MIN_REPAIR_SECONDS:
                log_ai_event("entry_expansion_repair_started", entry_type=entry_type, reason_code=repair_reason)
                output = self._complete_repair(
                    facts_text=facts_text,
                    primary_description=primary_description,
                    entry_type=entry_type,
                    context=context,
                    hard_required_facts=hard_required_facts,
                optimized="",
                    reason=repair_reason,
                    missing_hard=[],
                    started_at=started_at,
                )
                candidate = _normalize_candidate(output.optimized_description, entry_type)

        optimized, suggestions, warnings = partition_entry_text(candidate, fact_ledger)
        if not optimized and primary_description:
            optimized = primary_description
            warnings.append("candidate_contains_unconfirmed_additions")

        missing_hard = missing_hard_facts(hard_required_facts, optimized) if optimized else []
        if optimized and missing_hard:
            repair_reason = "hard_fact_omitted"
            log_ai_event(
                "entry_expansion_repair_started",
                entry_type=entry_type,
                reason_code=repair_reason,
                hard_fact_count=len(hard_required_facts),
                omitted_fact_count=len(missing_hard),
            )
            optimized, extra_suggestions, extra_warnings, output = self._repair_material_omissions(
                optimized,
                missing_hard,
                fact_ledger,
                facts_text=facts_text,
                primary_description=primary_description,
                entry_type=entry_type,
                context=context,
                hard_required_facts=hard_required_facts,
                reason=repair_reason,
                previous_output=output,
                started_at=started_at,
            )
            suggestions.extend(extra_suggestions)
            warnings.extend(extra_warnings)

        if not optimized:
            fallback_reason = "repair_failed" if repair_reason else "insufficient_facts"
            log_ai_event(
                "entry_expansion_rejected",
                level=logging.WARNING,
                entry_type=entry_type,
                reason_code=fallback_reason,
            )
            return {
                "optimized_description": "",
                "changes": [],
                "unconfirmed_suggestions": suggestions,
                "validation_warnings": list(dict.fromkeys(warnings)),
                "source": "ai_expanded",
                "generation_source": "llm",
                "fallback_reason": fallback_reason,
            }

        remaining_hard = missing_hard_facts(hard_required_facts, optimized)
        if remaining_hard:
            warnings.append("hard_fact_omitted_after_repair")
        return {
            "optimized_description": optimized,
            "changes": [],
            "unconfirmed_suggestions": suggestions[:6],
            "validation_warnings": list(dict.fromkeys(warnings)),
            "uncovered_facts": remaining_hard[:8],
            "source": "ai_expanded",
            "generation_source": "llm",
        }

    def _base_payload(
        self,
        *,
        facts_text: str,
        primary_description: str,
        entry_type: str,
        context: dict[str, Any],
        hard_required_facts: list[FactRequirement],
    ) -> dict[str, Any]:
        return {
            "entry_facts": facts_text,
            "primary_description": primary_description,
            "entry_type": entry_type or None,
            "target_position": context.get("target_position"),
            "instruction": context.get("instruction"),
            "protected_quantity_facts": quantified_fact_contexts(facts_text),
            "hard_required_facts": hard_required_facts,
        }

    def _complete_repair(
        self,
        *,
        facts_text: str,
        primary_description: str,
        entry_type: str,
        context: dict[str, Any],
        hard_required_facts: list[FactRequirement],
        optimized: str,
        reason: str,
        missing_hard: list[str],
        started_at: float,
    ) -> EntryExpansionOutput:
        payload = self._base_payload(
            facts_text=facts_text,
            primary_description=primary_description,
            entry_type=entry_type,
            context=context,
            hard_required_facts=hard_required_facts,
        )
        payload.update({
            "rejected_candidate": optimized,
            "rejected_reason": reason,
            "omitted_facts": missing_hard,
        })
        return self._complete(
            schema_name="entry_expansion_repair",
            system_prompt=_repair_prompt(entry_type),
            payload=payload,
            remaining_seconds=self._remaining_seconds(started_at),
        )

    def _repair_material_omissions(
        self,
        optimized: str,
        missing_hard: list[str],
        fact_ledger: list[dict[str, str]],
        *,
        facts_text: str,
        primary_description: str,
        entry_type: str,
        context: dict[str, Any],
        hard_required_facts: list[FactRequirement],
        reason: str,
        previous_output: EntryExpansionOutput,
        started_at: float,
    ) -> tuple[str, list[str], list[str], EntryExpansionOutput]:
        """Run at most one repair pass; semantic source text is never raw-appended."""
        remaining_seconds = self._remaining_seconds(started_at)
        if remaining_seconds is not None and remaining_seconds < self._MIN_REPAIR_SECONDS:
            return optimized, [], ["repair_skipped_budget"], previous_output
        try:
            repaired = self._complete_repair(
                facts_text=facts_text,
                primary_description=primary_description,
                entry_type=entry_type,
                context=context,
                hard_required_facts=hard_required_facts,
                optimized=optimized,
                reason=reason,
                missing_hard=missing_hard,
                started_at=started_at,
            )
        except Exception as exc:
            log_ai_event(
                "entry_expansion_coverage_repair_failed",
                level=logging.WARNING,
                entry_type=entry_type,
                reason_code=getattr(exc, "reason_code", type(exc).__name__),
            )
            return optimized, [], ["repair_failed"], EntryExpansionOutput(
                optimized_description=optimized,
            )
        repaired_text, extra_suggestions, repair_warnings = partition_entry_text(
            _normalize_candidate(repaired.optimized_description, entry_type), fact_ledger
        )
        if not repaired_text:
            return optimized, [], ["repair_failed"], previous_output
        preserved_initial = [
            fact for fact in hard_required_facts if hard_fact_is_preserved(fact, optimized)
        ]
        repaired_missing = missing_hard_facts(hard_required_facts, repaired_text)
        if (
            len(repaired_missing) >= len(missing_hard)
            or any(not hard_fact_is_preserved(fact, repaired_text) for fact in preserved_initial)
            or _repair_regresses_structure(optimized, repaired_text)
        ):
            return optimized, [], ["repair_rejected_quality_regression"], previous_output
        return repaired_text, extra_suggestions, repair_warnings, repaired

    def _remaining_seconds(self, started_at: float) -> float | None:
        if self.timeout_seconds is None:
            return None
        return max(0.1, self.timeout_seconds - (time.perf_counter() - started_at))

    def _complete(
        self, *, schema_name: str, system_prompt: str, payload: dict[str, Any], remaining_seconds: float | None
    ) -> EntryExpansionOutput:
        kwargs: dict[str, Any] = {
            "schema": EntryExpansionOutput,
            "schema_name": schema_name,
            "system_prompt": system_prompt,
            "payload": payload,
        }
        if remaining_seconds is not None:
            kwargs.update(timeout_seconds=remaining_seconds, max_attempts=1)
        return self.completion.complete(**kwargs)


def _repair_regresses_structure(original: str, repaired: str) -> bool:
    original_lines = [line for line in original.splitlines() if line.strip()]
    repaired_lines = [line for line in repaired.splitlines() if line.strip()]
    if len(original_lines) >= 2 and len(repaired_lines) < len(original_lines):
        return True
    return len(original) >= 120 and len(repaired) < len(original) * 0.65

def _normalize_candidate(candidate: str, entry_type: str) -> str:
    text = candidate.strip()
    if not text or entry_type == "education":
        return text
    return normalize_bullet_description(text)


class FallbackEntryExpander:
    def __init__(self, primary: EntryExpander, fallback: EntryExpander) -> None:
        self.primary = primary
        self.fallback = fallback

    def expand(self, entry: dict[str, Any], *, context: dict[str, Any]) -> dict[str, Any]:
        try:
            return self.primary.expand(entry, context=context)
        except Exception as exc:
            reason = exc.reason_code if isinstance(exc, LLMServiceError) else type(exc).__name__.lower()[:48]
            log_ai_event(
                "entry_expansion_failed",
                level=logging.ERROR,
                entry_type=str(context.get("entry_type") or ""),
                reason_code=reason,
                trace_id=getattr(exc, "trace_id", None),
                exception=type(exc).__name__,
            )
            fallback = self.fallback.expand(entry, context=context)
            optimized = str(fallback.get("optimized_description") or "").strip()
            if optimized:
                return {
                    **fallback,
                    "source": str(fallback.get("source") or "rule_polish"),
                    "generation_source": "rule_fallback",
                    "fallback_reason": reason,
                }
            return {
                "optimized_description": "",
                "changes": [],
                "unconfirmed_suggestions": [],
                "source": "rule_polish",
                "generation_source": "unavailable",
                "fallback_reason": reason,
            }


def _entry_facts(entry: dict[str, Any]) -> str:
    return "\n".join(item["text"] for item in _entry_fact_ledger(entry))


def _entry_fact_ledger(entry: dict[str, Any]) -> list[dict[str, str]]:
    keys = (
        "title", "organization", "role", "company", "position", "project_name", "project_role",
        "school", "major", "degree", "start_date", "end_date_or_present", "name", "award", "date",
        "description",
    )
    ledger: list[dict[str, str]] = []
    for key in keys:
        value = str(entry.get(key) or "").strip()
        if value:
            ledger.append({"id": f"entry_{key}", "field": key, "text": value})
    for index, value in enumerate(entry.get("highlights") or [], start=1):
        clean = str(value).strip()
        if clean:
            ledger.append({"id": f"entry_highlight_{index}", "field": "highlight", "text": clean})
    return ledger


def build_expander(settings: Settings, client: Any | None = None) -> EntryExpander:
    rules = RuleBasedEntryExpander()
    if not settings.use_openai:
        return rules
    completion = OpenAICompatibleStructuredClient(settings, client)
    primary = OpenAIEntryExpander(completion)
    return FallbackEntryExpander(primary, rules) if settings.fallback_to_rules else primary