Blame for app/services/proposal_generator.py

"""Proposal Generator.

Drafts a grounded Upwork proposal *only* from the evidence index. The
generator never invents past clients, projects, metrics,
certifications, tools, or outcomes — every factual claim it emits is
tied back to a proof point's ``evidence_id``.

The AI writer revises for relevance, structure and natural voice before the
fact-check. Deterministic generation is an explicitly enabled development
fallback, not a client-ready replacement for the AI writer.

Output contract::

    {
        "proposal": str,                # the drafted text
        "factual_claims": list[dict],   # each claim has an evidence_id
        "target_length": (min, max),    # word count band
        "complexity": str,              # "simple" | "standard" | "complex"
        "word_count": int,
    }
"""

from __future__ import annotations

import inspect
import json
import re
from typing import Any, Iterable, Optional

from app.config import get_settings
from app.prompts.proposal_prompt import FORBIDDEN_PHRASES
from app.prompts.proposal_prompt import render_prompt as render_proposal_prompt
from app.prompts.system_prompt import SYSTEM_PROMPT
from app.services import llm_client
from app.services.match_engine import _is_missing as is_missing
from app.services.verification import (
    _plain_text as plain_text,
    strip_evidence_ids,
    verify as run_verification,
)


TASK_NAME = "proposal_generation"


# Default user-facing error message for size-related API failures. The
# raw provider error message (already sanitized of organization IDs) is
# attached separately via ``ProposalGenerationError.sanitized_error``.
SIZE_FAILURE_USER_MESSAGE = (
    "Proposal generation failed because the request was too large for "
    "the current API token limit. The app reduced context and retried "
    "once, but the API still rejected the request."
)

# The reply, not the request, was too long: the model ran out of output
# room before it finished, even after a retry with more room.
TRUNCATION_FAILURE_USER_MESSAGE = (
    "Proposal generation failed because the AI's reply was cut off by the "
    "output token limit before it finished. Raise 'Proposal max output "
    "tokens' under Advanced options on the Setup screen and try again."
)


# Claim types we will allow into the compact proposal context, in
# preference order. Anything outside this whitelist is treated as low
# value and dropped before the LLM call.
PREFERRED_CLAIM_TYPES: tuple[str, ...] = (
    "positioning",
    "selected_offer",
    "service",
    "deliverable",
    "skill",
    "tool",
    "industry",
    "project",
    "work_history",
    "metric",
    "testimonial",
    "achievement",
    "certification",
    "proposal_preference",
)


_CONFIDENCE_RANK = {"high": 3, "medium": 2, "low": 1}


# Cap on a single claim's text inside the compact context. Long
# free-text claim bodies are what blow the request up.
_CLAIM_TEXT_CAP_DEFAULT = 240
_CLAIM_TEXT_CAP_SMALL = 160


class ProposalGenerationError(RuntimeError):
    """Raised when proposal generation cannot be completed.

    The UI catches this so the user sees a clear failure message rather
    than a silent fake proposal.

    ``sanitized_error`` is the provider-reported reason with
    organization IDs and other private metadata stripped. ``meta`` is a
    small dict of context the UI may surface (provider, model, evidence
    points sent, approximate context size, retry used).
    """

    def __init__(
        self,
        message: str,
        *,
        status: str = "failed",
        llm_result: Any = None,
        sanitized_error: Optional[str] = None,
        meta: Optional[dict] = None,
    ):
        super().__init__(message)
        self.status = status
        self.llm_result = llm_result
        self.sanitized_error = sanitized_error
        self.meta = meta or {}


LENGTH_BANDS: dict[str, tuple[int, int]] = {
    "simple": (150, 250),
    "standard": (250, 400),
    "complex": (350, 550),
}


# FORBIDDEN_PHRASES is the single source of truth in app.prompts.proposal_prompt
# (Instruction Set 2). Imported above and reused here for output stripping so
# the prompt and the post-processing never drift apart.


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _field(confirmed_job: dict, key: str) -> str:
    entry = (confirmed_job or {}).get(key) or {}
    if isinstance(entry, dict):
        return str(entry.get("value", "") or "").strip()
    return str(entry or "").strip()


def _present(value: str) -> Optional[str]:
    return None if is_missing(value) else value


def _proof_attr(proof: Any, name: str, default: Any = None) -> Any:
    if hasattr(proof, name):
        return getattr(proof, name)
    if isinstance(proof, dict):
        return proof.get(name, default)
    return default


def _proofs_by_type(
    evidence: Iterable, claim_types: set[str]
) -> list:
    return [p for p in evidence if _proof_attr(p, "claim_type") in claim_types]


def _word_count(text: str) -> int:
    return len(re.findall(r"\b\w[\w'-]*\b", text))


def _detect_complexity(confirmed_job: dict) -> str:
    description = _field(confirmed_job, "job_description")
    duration = _field(confirmed_job, "project_duration").lower()
    experience_level = _field(confirmed_job, "experience_level").lower()

    description_len = _word_count(description) if not is_missing(description) else 0

    if "expert" in experience_level or "more than 6" in duration or description_len > 200:
        return "complex"
    if "entry" in experience_level or "less than 1" in duration or description_len < 50:
        return "simple"
    return "standard"


# Headings and labels from the team's coaching documents. They mean nothing
# to a client, so any that slip into a draft (from the model or the local
# outline) are removed: "YOUR SELECTED OFFER", "PART A — YOUR OFFER
# STATEMENT", "HOW THIS HELPS THE CLIENT", "Offer 2 — ".
_INTERNAL_LABELS = re.compile(
    r"\byour selected offer\b[\s,:—–-]*|\bpart [ab]\s*[—–:-]\s*|"
    r"\byour offer statement\b[\s,:—–-]*",
    re.IGNORECASE,
)
# "Offer 2 —" and "how this helps the client" are also ordinary prose ("I can
# offer 2–3 concepts", "here is how this helps the client team"), so these
# two count as a label only when they open a line.
_LINE_START_LABELS = re.compile(
    r"^([ \t]*)(?:how this helps the (?:client|brand)\b[ \t]*(?:[:—–-][ \t]*|$)|"
    r"(?:offer|niche)[ \t]*\d+[ \t]*[a-z]?[ \t]*[—–:][ \t]*)([a-z]?)",
    re.IGNORECASE | re.MULTILINE,
)
# Four or more capitalised words on a line of their own ("ACADEMIC WRITING
# SUPPORT FOR MANAGEMENT") are a pasted document heading. Inside a sentence
# the same run is content ("fluent in HTML CSS PHP SQL"), and so is a
# client's hidden keyword written in capitals, so neither is touched.
_CAPS_HEADING_LINE = re.compile(
    r"^[ \t]*(?:[A-Z][A-Z&/'’-]*[ \t]+){3,}[A-Z][A-Z&/'’-]*[ \t]*[:—–-]?[ \t]*(?:\n|$)",
    re.MULTILINE,
)

# Layout and formatting tags a model sometimes leaks into its reply ("</div>",
# "<br/>", "<p>", "<strong>"). Only these are stripped, and only in lower
# case: a proposal for a developer job may quote code ("Map<String,
# List<Order>>", "<Suspense>", "<Table>", "React.FC<P>") or a comparison
# ("< 2%"), and that text must survive. Attributes need a value ('<div
# class="x">'), so prose such as "<b and c>" or "<a third" is never taken
# for a tag, and nothing is matched across a line break.
_HTML_ATTR = r"""(?:[ \t]+[A-Za-z_:][-\w:.]*[ \t]*=[ \t]*(?:"[^"\n]*"|'[^'\n]*'|[^\s"'<>=`]+))"""
_STRAY_HTML_TAG = re.compile(
    r"</(?:div|span|p|br|hr|b|i|u|strong|em|ul|ol|li|h[1-6]|font|section|table|tr|td|th|a)[ \t]*>"
    r"|<(?:div|span|p|br|hr|b|i|u|strong|em|ul|ol|li|h[1-6]|font|section|table|tr|td|th)"
    + _HTML_ATTR + r"*[ \t]*/?>"
    r"|<a" + _HTML_ATTR + r"+[ \t]*/?>"
)

# Plain words for the banned buzzwords, by inflected form, so a banned word
# is swapped on word boundaries ("leveraged" -> "used") instead of being cut
# out of its sentence, or out of a longer word ("seamlessly" -> "ly"). An
# empty swap drops a lead-in whose sentence stands without it. A banned
# phrase with no entry here is a filler sentence ("I'd be happy to help…")
# and is dropped whole; a banned single word with no entry is dropped.
_BANNED_SWAPS: dict[str, dict[str, str]] = {
    "leverage": {"leverage": "use", "leverages": "uses", "leveraged": "used", "leveraging": "using"},
    "utilize": {
        "utilize": "use", "utilizes": "uses", "utilized": "used", "utilizing": "using",
        "utilization": "use", "utilise": "use", "utilises": "uses", "utilised": "used",
        "utilising": "using", "utilisation": "use",
    },
    "seamless": {"seamless": "smooth", "seamlessly": "smoothly"},
    "robust": {"robust": "reliable", "robustly": "reliably", "robustness": "reliability"},
    "game-changer": {"game-changer": "big win", "game-changers": "big wins"},
    "cutting-edge": {"cutting-edge": "modern"},
    "delve": {"delve": "dig", "delves": "digs", "delved": "dug", "delving": "digging"},
    "straightforward": {"straightforward": "simple", "straightforwardly": "simply"},
    "i am confident that": {"i am confident that": ""},
    "i am passionate about": {"i am passionate about": "I care about"},
}
# Marks where a lead-in was dropped, so the sentence after it can be
# capitalised; removed before the text is returned.
_DROPPED = "\x00"

# Number, day and month ranges ("24–48 hours", "$500–$800", "9am–5pm",
# "Monday–Friday") keep their meaning as a hyphen. An en dash with or
# without spaces, or an em dash with none, marks a range; a spaced em dash
# between numbers is a pause and is left to the comma rule.
_RANGE_DASH = r"(?:[ \t]*–[ \t]*|—)"
_NUMBER_RANGE = re.compile(r"(\d[%a-zA-Z]{0,3})" + _RANGE_DASH + r"([$€£₹]?[A-Za-z]{0,2}\d)")
_DAY_OR_MONTH = (
    r"(?:(?:mon|tues|wednes|thurs|fri|satur|sun)day|mon|tue|wed|thu|fri|sat|sun|"
    r"january|february|march|april|may|june|july|august|september|october|"
    r"november|december|jan|feb|mar|apr|jun|jul|aug|sept?|oct|nov|dec)"
)
_NAMED_RANGE = re.compile(
    r"\b(" + _DAY_OR_MONTH + r")\.?" + _RANGE_DASH + r"(" + _DAY_OR_MONTH + r"\b)",
    re.IGNORECASE,
)
# The step template reads "Step 1: Audit — Review …": the label becomes a
# sentence of its own ("Step 1: Audit. Review …"), not a comma splice.
_STEP_LABEL_DASH = re.compile(
    r"^([ \t]*Step[ \t]+\d+[ \t]*:[ \t]*[^\n—–.!?:]{1,60}?)[ \t]*[—–][ \t]*(\w)",
    re.MULTILINE,
)


def strip_stray_html(text: str) -> str:
    """Remove leaked layout/formatting HTML tags, and only those.

    Proposals are plain prose, so a stray ``</div>`` must not reach the
    copied / downloaded text; anything else in angle brackets is content.
    """
    if not text:
        return text
    return _STRAY_HTML_TAG.sub("", text)


def _match_case(original: str, plain: str) -> str:
    if len(original) > 1 and original.isupper():
        return plain.upper()
    if original[:1].isupper():
        return plain[:1].upper() + plain[1:]
    return plain


def _phrase_regex(phrase: str) -> str:
    """Regex for ``phrase`` with flexible spacing and either apostrophe."""
    return r"[ \t]+".join(
        re.escape(word).replace("'", "['’]") for word in phrase.split()
    )


def _swap_banned(text: str, phrase: str) -> str:
    """Replace or drop one banned word or phrase on word boundaries."""
    swaps = _BANNED_SWAPS.get(phrase.lower())
    if swaps is None and len(phrase.split()) > 1:
        # A filler sentence: drop the sentence that holds it, so no
        # fragment such as "with the Shopify migration." is left behind.
        pattern = re.compile(r"(?<!\w)" + _phrase_regex(phrase) + r"(?!\w)", re.IGNORECASE)
        if not pattern.search(text):
            return text
        lines: list[str] = []
        for line in text.split("\n"):
            if pattern.search(line):
                kept = [s for s in re.split(r"(?<=[.!?])[ \t]+", line) if not pattern.search(s)]
                line = " ".join(kept)
                if not line.strip():
                    continue
            lines.append(line)
        return "\n".join(lines)
    swaps = swaps or {phrase.lower(): ""}
    forms = sorted(swaps, key=len, reverse=True)
    # A banned word inside a hyphenated name, a link or a handle
    # ("robust-tools.com", "@seamless") is part of that name and stays.
    pattern = re.compile(
        r"(?<![\w/.@-])(?:" + "|".join(_phrase_regex(f) for f in forms) + r")(?![\w-]|[./@]\w)",
        re.IGNORECASE,
    )

    def _plain(match: re.Match) -> str:
        found = match.group(0)
        key = " ".join(found.replace("’", "'").split()).lower()
        plain = swaps.get(key, "")
        return _match_case(found, plain) if plain else _DROPPED

    return pattern.sub(_plain, text)


def _strip_forbidden(text: str, *, hidden_keyword: str = "") -> str:
    """Clean a draft for the client: leaked tags, internal labels, pasted
    headings, banned words and phrases, and em dashes.

    ``hidden_keyword`` is the client's keyword; a heading-like line that
    holds it is kept.
    """
    # Strip leaked HTML tags (e.g. a trailing </div>) so the copied /
    # downloaded text is clean too, not just the on-screen render.
    text = strip_stray_html(text)
    # Ranges first, so "offer 2–3" is not read as an "Offer 2 —" label.
    text = _NUMBER_RANGE.sub(r"\1-\2", text)
    text = _NAMED_RANGE.sub(r"\1-\2", text)
    text = _STEP_LABEL_DASH.sub(lambda m: m.group(1) + ". " + m.group(2).upper(), text)
    text = _INTERNAL_LABELS.sub("", text)
    text = _LINE_START_LABELS.sub(lambda m: m.group(1) + m.group(2).upper(), text)
    keyword = " ".join((_present(hidden_keyword or "") or "").split()).lower()

    def _heading(match: re.Match) -> str:
        line = " ".join(match.group(0).split()).strip(" :—–-").lower()
        if keyword and (keyword in line or line in keyword):
            return match.group(0)
        return ""

    text = _CAPS_HEADING_LINE.sub(_heading, text)
    for phrase in FORBIDDEN_PHRASES:
        text = _swap_banned(text, phrase)
    # A sentence that opened with a dropped lead-in opens with the next word.
    text = re.sub(
        r"((?:^|[.!?])[ \t]*)" + _DROPPED + r"[ \t]*([a-z])",
        lambda m: m.group(1) + m.group(2).upper(),
        text,
        flags=re.MULTILINE,
    )
    text = re.sub(_DROPPED + r"[ \t]*", "", text)
    # Instruction Set 2 bans em dashes — convert any other em/en dash to a
    # comma so the rule holds even if the model slips one in. A dash that
    # opens a line is a bullet mark and goes; no dash joins two lines.
    text = re.sub(r"^([ \t]*)[—–][ \t]*", r"\1", text, flags=re.MULTILINE)
    text = re.sub(r"[ \t]*[—–][ \t]*", ", ", text)
    text = re.sub(r"[ \t]{2,}", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


# The end of a sentence: terminal punctuation, maybe a closing quote or
# bracket, then a space, a line break or the end of the text.
_SENTENCE_END = re.compile(r"[.!?…][\"'’”)\]]*(?=\s|$)")


def _trim_to_band(text: str, band: tuple[int, int]) -> str:
    _, max_words = band
    words = list(re.finditer(r"\S+", text))
    if len(words) <= max_words:
        return text
    # Trim from the soft CTA backward, never from credibility/understanding.
    # The draft is cut after its last whole sentence or line within the
    # band, so its paragraph breaks and step lines survive as written.
    limit = words[max_words - 1].end()
    cuts = [m.end() for m in _SENTENCE_END.finditer(text, 0, limit)]
    cuts += [m.start() for m in re.finditer(r"\n", text[:limit])]
    trimmed = text[: max(cuts)].rstrip() if cuts else ""
    if not trimmed:
        # Prefer any complete sentence over a fragment padded to a minimum.
        # A single overlong sentence still needs a bounded word-level cut.
        trimmed = text[:limit]
    if not re.search(r"[.!?…][\"'’”)\]]*$", trimmed):
        trimmed = trimmed.rstrip(",;:—–- \t\n") + "."
    return trimmed


# ---------------------------------------------------------------------------
# Citation helper
# ---------------------------------------------------------------------------


def _claim(text: str, proof: Any, *, kind: str) -> dict[str, Any]:
    return {
        "text": text,
        "kind": kind,
        "evidence_id": _proof_attr(proof, "evidence_id"),
        "source_file": _proof_attr(proof, "source_file"),
        "claim_type": _proof_attr(proof, "claim_type"),
    }


def _bare_claim(text: str, kind: str) -> dict[str, Any]:
    return {
        "text": text,
        "kind": kind,
        "evidence_id": None,
        "source_file": None,
        "claim_type": None,
    }


# ---------------------------------------------------------------------------
# Section builders
# ---------------------------------------------------------------------------


def _opening(confirmed_job: dict, claims: list[dict]) -> str:
    title = _present(_field(confirmed_job, "job_title"))
    client_need = _present(_field(confirmed_job, "client_need"))
    if title and client_need:
        return f"Saw your {title} post — the priority of {client_need.lower()} matches what I work on."
    if title:
        return f"Quick note on your {title} post — sharing how I'd approach it."
    if client_need:
        return f"On your stated goal — {client_need} — here is how I would approach it."
    return "Here is how I'd approach the brief you've shared."


_SECOND_PERSON = re.compile(r"\s(?:You|Your)\b")
# A document heading glued to the front of a dossier paragraph ("ACADEMIC
# WRITING SUPPORT FOR MANAGEMENT Natural Communication…") is not part of
# its title.
_GLUED_HEADING = re.compile(r"^(?:[A-Z][A-Z&/'’-]+\s+){3,}[A-Z][A-Z&/'’-]+\s+(?=[A-Z][a-z])")


def _lead_phrase(text: str) -> str:
    """The short title of a coaching-document paragraph.

    Dossier strengths read "Natural Communication Confidence You described
    communication as your forte…": the title is the part a client may see;
    the rest is written to the student. Text without such a title is cut to
    its first sentence.
    """
    text = _GLUED_HEADING.sub("", " ".join(str(text or "").split()))
    match = _SECOND_PERSON.search(text)
    if match and 8 <= match.start() <= 120:
        return text[: match.start()].strip(" .,:;—-")
    sentence = re.split(r"(?<=[.!?])\s", text, maxsplit=1)[0]
    return sentence[:160].strip()


def _credibility(evidence: list, matched_skills: list[str], claims: list[dict]) -> str:
    parts: list[str] = []
    # Strength titles first: they are the dossier's own summary of what the
    # freelancer brings, and they read cleanly to a client.
    strengths = [
        _lead_phrase(_proof_attr(p, "claim_text", ""))
        for p in _proofs_by_type(evidence, {"achievement"})
    ]
    # A paragraph that starts in lower case is the tail of one cut by a page
    # break, not a strength of its own.
    strengths = [s for s in strengths if s and s[0].isupper()][:3]
    if strengths:
        parts.append("What I bring: " + "; ".join(strengths) + ".")
    experience_proofs = _proofs_by_type(evidence, {"experience", "work_history"})
    if experience_proofs and not strengths:
        proof = experience_proofs[0]
        text = _lead_phrase(_proof_attr(proof, "claim_text", ""))
        if text:
            parts.append(text)
            claims.append(_claim(text, proof, kind="experience"))

    skill_proofs = _proofs_by_type(evidence, {"skill", "tool"})
    cited_skills: list[str] = []
    for proof in skill_proofs:
        for skill in (_proof_attr(proof, "skills", []) or []) + (
            _proof_attr(proof, "tools", []) or []
        ):
            if skill and skill.lower() in {s.lower() for s in matched_skills}:
                if skill not in cited_skills:
                    cited_skills.append(skill)
                    claims.append(_claim(skill, proof, kind="skill"))
        if len(cited_skills) >= 4:
            break
    if cited_skills:
        parts.append("Hands-on with " + ", ".join(cited_skills[:4]) + ".")

    metrics = _proofs_by_type(evidence, {"metric"})
    if metrics:
        proof = metrics[0]
        text = _lead_phrase(_proof_attr(proof, "claim_text", ""))
        if text:
            parts.append(text)
            claims.append(_claim(text, proof, kind="achievement"))

    if not parts:
        return "I'll lean on approach here — dossier evidence for this brief is light."
    return " ".join(parts)


def _understanding(confirmed_job: dict) -> str:
    need = _present(_field(confirmed_job, "client_need"))
    deliverables = _present(_field(confirmed_job, "required_deliverables"))
    description = _present(_field(confirmed_job, "job_description"))
    if need and deliverables:
        return f"As I read it, the goal is {need}, anchored on {deliverables}."
    if need:
        return f"As I read it, the goal is {need}."
    if description:
        return f"Reading the brief, the priority looks like: {description[:140]}"
    return "Happy to confirm the exact priority once you share more context."


def _approach(evidence: list, claims: list[dict]) -> str:
    tools = _proofs_by_type(evidence, {"tool"})
    services = _proofs_by_type(evidence, {"service", "deliverable"})
    bullets: list[str] = []

    if services:
        for proof in services[:2]:
            # A dossier deliverable reads "1. Content Audit You review every
            # page…": its title is the step, the rest is written to the student.
            text = re.sub(r"^\d+\\?[.)]\s*", "", _lead_phrase(_proof_attr(proof, "claim_text", "")))
            if text:
                bullets.append(text)
                claims.append(_claim(text, proof, kind="service"))

    if tools:
        proof = tools[0]
        tool_list = _proof_attr(proof, "tools", []) or []
        if tool_list:
            phrase = "Stack I'd use: " + ", ".join(tool_list[:5])
            bullets.append(phrase)
            claims.append(_claim(phrase, proof, kind="tool"))

    if not bullets:
        bullets.append("Step 1: confirm scope and success criteria with you.")
        bullets.append("Step 2: ship a small milestone you can review before we go wide.")

    return "How I'd approach it:\n- " + "\n- ".join(bullets[:3])


def _differentiator(evidence: list, claims: list[dict]) -> str:
    # The offer statement is not quoted here: it is the angle, not proof, and
    # its numbers are promises ("increase conversion by 40-80%").
    portfolio = _proofs_by_type(evidence, {"portfolio", "testimonial", "project"})
    if portfolio:
        proof = portfolio[0]
        text = _lead_phrase(_proof_attr(proof, "claim_text", ""))
        if text:
            claims.append(_claim(text, proof, kind="portfolio"))
            return f"For context on past work: {text}"
    return "Why me: I scope tightly and only commit to outcomes I can evidence."


def _smart_question(confirmed_job: dict, match_data: Optional[dict]) -> str:
    if is_missing(_field(confirmed_job, "required_deliverables")):
        return "One question: what does done look like for the first milestone?"
    if is_missing(_field(confirmed_job, "budget_or_rate")):
        return "One question: is there a target budget shape (fixed vs. hourly) you're anchored on?"
    if is_missing(_field(confirmed_job, "project_duration")):
        return "One question: what timeline are you aiming for on phase 1?"
    if match_data and match_data.get("missing_critical_fields"):
        missing = match_data["missing_critical_fields"][0]
        nice = missing.replace("_", " ")
        return f"One question to align: could you share more on {nice}?"
    return "One question: what would make this engagement feel like a clear win three months in?"


def _soft_cta() -> str:
    return "Happy to walk through how I'd scope phase 1 if useful."


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------


def _local_generate(
    confirmed_job: dict,
    evidence_list: list,
    matched_skills: list[str],
    band: tuple[int, int],
) -> tuple[str, list[dict[str, Any]]]:
    """Run the deterministic local section assembler."""
    claims: list[dict[str, Any]] = []
    sections = [
        _opening(confirmed_job, claims),
        _credibility(evidence_list, matched_skills, claims),
        _understanding(confirmed_job),
        _approach(evidence_list, claims),
        _differentiator(evidence_list, claims),
        _smart_question(confirmed_job, None),
        _soft_cta(),
    ]
    proposal = "\n\n".join(s for s in sections if s)
    proposal = _strip_forbidden(proposal)
    proposal = _trim_to_band(proposal, band)
    return proposal, claims


# The size retry's cap on one job field. A post states its ask at the top
# and often its instructions ("mention a similar project") at the bottom,
# so a long field loses its middle.
_COMPACT_JOB_FIELD_CAPS = {"job_description": 1000}
_COMPACT_JOB_FIELD_CAP = 400


def _clip_middle(text: str, cap: int) -> str:
    """``text`` cut to about ``cap`` characters, keeping its start and end."""
    if len(text) <= cap:
        return text
    head = cap * 2 // 3
    return text[:head].rstrip() + " … " + text[len(text) - (cap - head):].lstrip()


def _serialize_job_for_prompt(confirmed_job: dict, *, compact: bool = False) -> str:
    """The job post as the JSON block the writer prompt embeds.

    ``compact`` is for the size retry: only the fields the post shows, with
    long text cut down. The screening questions and hidden keyword are left
    out there because the prompt already gives each its own line.
    """
    if compact:
        fields = _compact_job_fields(confirmed_job)
        fields.pop("screening_questions", None)
        fields.pop("hidden_keyword", None)
        return json.dumps(
            {
                key: _clip_middle(
                    value, _COMPACT_JOB_FIELD_CAPS.get(key, _COMPACT_JOB_FIELD_CAP)
                )
                for key, value in fields.items()
            },
            ensure_ascii=False,
            indent=2,
        )
    return json.dumps(_compact_job_fields(confirmed_job), ensure_ascii=False, indent=2)


def _serialize_evidence_for_prompt(evidence_list: list, *, claim_text_cap: int) -> str:
    """Serialize the compact evidence subset into the JSON block that the
    proposal prompt embeds. The full proof point — and the source file
    that holds it — is never serialized; only the evidence_id, source
    type, claim type, and a truncated claim text are sent."""
    out: list[dict[str, Any]] = []
    for proof in evidence_list:
        out.append(
            {
                "evidence_id": _proof_attr(proof, "evidence_id"),
                "source_type": _proof_attr(proof, "source_type"),
                "claim_type": _proof_attr(proof, "claim_type"),
                "claim": str(_proof_attr(proof, "claim_text", "") or "").strip()[
                    :claim_text_cap
                ],
            }
        )
    return json.dumps(out, ensure_ascii=False, indent=2)


# ---------------------------------------------------------------------------
# Compact proposal context builder
# ---------------------------------------------------------------------------


def _tokens_from(value: str) -> set[str]:
    return {
        t.lower()
        for t in re.findall(r"[a-zA-Z][a-zA-Z0-9+.#-]{2,}", value or "")
        if t
    }


def _job_signal_tokens(confirmed_job: dict) -> set[str]:
    tokens: set[str] = set()
    for key in (
        "job_title",
        "job_description",
        "client_need",
        "required_deliverables",
        "required_skills",
    ):
        tokens |= _tokens_from(_field(confirmed_job, key))
    return tokens


def _score_proof(
    proof: Any,
    *,
    job_tokens: set[str],
    matched_skills_norm: set[str],
    matched_industries_norm: set[str],
    referenced_ids: set[str],
) -> float:
    """Score a proof point for relevance to the current job posting."""
    score = 0.0

    claim_type = (_proof_attr(proof, "claim_type") or "").lower()
    if claim_type in PREFERRED_CLAIM_TYPES:
        # Earlier entries in the preference list get a small extra nudge.
        rank = PREFERRED_CLAIM_TYPES.index(claim_type)
        score += 6.0 + (len(PREFERRED_CLAIM_TYPES) - rank) * 0.15
    elif claim_type == "other_relevant_evidence":
        score -= 2.0

    # Lower source_priority = higher quality source.
    source_priority = _proof_attr(proof, "source_priority", 17)
    try:
        score += max(0.0, 4.0 - (float(source_priority) / 5.0))
    except (TypeError, ValueError):
        pass

    confidence = (_proof_attr(proof, "confidence") or "low").lower()
    score += _CONFIDENCE_RANK.get(confidence, 1)

    # If the match engine had already flagged this proof point, prefer it.
    if _proof_attr(proof, "evidence_id") in referenced_ids:
        score += 5.0

    # Overlap with matched skills / industries that the match engine
    # already proved are relevant to this job.
    skill_overlap = 0
    for entry in _proof_attr(proof, "skills", []) or []:
        if str(entry).strip().lower() in matched_skills_norm:
            skill_overlap += 1
    for entry in _proof_attr(proof, "tools", []) or []:
        if str(entry).strip().lower() in matched_skills_norm:
            skill_overlap += 1
    score += min(skill_overlap, 4) * 1.5

    industry_overlap = 0
    for entry in _proof_attr(proof, "industries", []) or []:
        if str(entry).strip().lower() in matched_industries_norm:
            industry_overlap += 1
    score += min(industry_overlap, 3) * 1.0

    # Overlap with raw job text tokens.
    claim_text = str(_proof_attr(proof, "claim_text", "") or "").lower()
    if claim_text and job_tokens:
        claim_tokens = set(re.findall(r"[a-zA-Z][a-zA-Z0-9+.#-]{2,}", claim_text))
        claim_tokens = {t.lower() for t in claim_tokens}
        token_overlap = len(claim_tokens & job_tokens)
        score += min(token_overlap, 6) * 0.5

    # Penalize superseded conflicts.
    if _proof_attr(proof, "conflict_status") == "superseded":
        score -= 3.0

    return score


def _collect_referenced_evidence_ids(match_result: Any) -> set[str]:
    """Read evidence references from both current and legacy matcher results."""
    ids: set[str] = set()
    def visit(value):
        if isinstance(value, dict):
            for key, child in value.items():
                if key in {"evidence_ids", "referenced_evidence_ids", "evidence_ids_used", "matching_evidence_ids"}:
                    if isinstance(child, (list, tuple, set)):
                        ids.update(str(ev) for ev in child if ev)
                elif isinstance(child, (dict, list)):
                    visit(child)
        elif isinstance(value, list):
            for child in value:
                visit(child)
    visit(match_result)
    return ids


def _trim_claim_text(value: str, cap: int) -> str:
    text = (value or "").strip()
    if len(text) <= cap:
        return text
    return text[: max(cap - 1, 1)].rstrip() + "…"


def _shorten_claim_texts(proofs: list, cap: int) -> list:
    """Return shallow copies of proof points with claim_text truncated to ``cap``.

    Pydantic ``ProofPoint`` instances are not mutated. Dicts are copied.
    """
    shortened: list = []
    for proof in proofs:
        claim_text = str(_proof_attr(proof, "claim_text", "") or "")
        if len(claim_text) <= cap:
            shortened.append(proof)
            continue
        if hasattr(proof, "model_copy"):
            try:
                shortened.append(
                    proof.model_copy(update={"claim_text": _trim_claim_text(claim_text, cap)})
                )
                continue
            except Exception:  # noqa: BLE001
                pass
        if isinstance(proof, dict):
            copy = dict(proof)
            copy["claim_text"] = _trim_claim_text(claim_text, cap)
            shortened.append(copy)
        else:
            shortened.append(proof)
    return shortened


def _drop_lowest_confidence_first(proofs: list, target_count: int) -> list:
    """Trim the proof list down to ``target_count`` items.

    Lower-confidence and lower-source-priority proofs are dropped first.
    """
    if len(proofs) <= target_count:
        return list(proofs)
    indexed = list(enumerate(proofs))
    indexed.sort(
        key=lambda pair: (
            _CONFIDENCE_RANK.get(
                (_proof_attr(pair[1], "confidence") or "low").lower(), 1
            ),
            -float(_proof_attr(pair[1], "source_priority", 17) or 17),
            -pair[0],
        ),
        reverse=True,
    )
    keep = sorted(idx for idx, _ in indexed[:target_count])
    return [proofs[i] for i in keep]


def _select_relevant_proofs(
    evidence_index: Iterable,
    confirmed_job: dict,
    match_result: Optional[dict],
    *,
    max_points: int,
) -> list:
    """Return up to ``max_points`` proof points ranked by relevance."""
    proofs = [p for p in (evidence_index or [])]
    if not proofs:
        return []

    matched_skills_norm: set[str] = set()
    matched_industries_norm: set[str] = set()
    if isinstance(match_result, dict):
        for entry in (match_result.get("skill_match") or {}).get("matched") or []:
            matched_skills_norm.add(str(entry).strip().lower())
        for entry in (match_result.get("industry_match") or {}).get("matched") or []:
            matched_industries_norm.add(str(entry).strip().lower())

    referenced_ids = _collect_referenced_evidence_ids(match_result)
    job_tokens = _job_signal_tokens(confirmed_job)

    scored = sorted(
        proofs,
        key=lambda p: _score_proof(
            p,
            job_tokens=job_tokens,
            matched_skills_norm=matched_skills_norm,
            matched_industries_norm=matched_industries_norm,
            referenced_ids=referenced_ids,
        ),
        reverse=True,
    )

    unique: list = []
    seen_text: set[str] = set()
    for proof in scored:
        key = re.sub(r"\W+", " ", str(_proof_attr(proof, "claim_text", ""))).strip().casefold()
        if key and key in seen_text:
            continue
        seen_text.add(key)
        unique.append(proof)
    substantive = {"project", "work_history", "metric", "testimonial", "achievement", "certification"}
    reserved = []
    seen_types = set()
    for proof in unique:
        kind = _proof_attr(proof, "claim_type")
        if kind in substantive and kind not in seen_types:
            reserved.append(proof)
            seen_types.add(kind)
    # At most half the slots are reserved; relevance fills the remainder.
    reserved = reserved[:max(1, max_points // 2)] if max_points > 0 else []
    selected: list = []
    seen_ids: set[str] = set()
    for proof in reserved + unique:
        if len(selected) >= max_points:
            break
        ev_id = _proof_attr(proof, "evidence_id")
        if (ev_id and ev_id in seen_ids) or any(proof is item for item in selected):
            continue
        if ev_id:
            seen_ids.add(ev_id)
        selected.append(proof)
    return selected


def build_proposal_context(
    confirmed_job_fields: dict,
    evidence_index: Iterable,
    match_result: Optional[dict] = None,
    recommendation_result: Optional[dict] = None,
    max_evidence_points: int = 20,
    max_context_chars: int = 15000,
) -> dict[str, Any]:
    """Build the compact context that the proposal LLM call will see.

    The function only ever returns metadata-shaped data:

    * confirmed job fields (already small)
    * recommendation verdict + best proposal angle
    * top-N relevant evidence points (no source_file leakage, no full
      dossier text)
    * matched skills / tools / industries / services from the match
      engine
    * the selected offer if relevant
    * pricing only if pricing surfaced as relevant for this job
    * missing-info warnings from the match engine and recommendation

    The full evidence index, raw dossier chunks, transcript text, or
    PDF text is never embedded.
    """
    confirmed_job_fields = confirmed_job_fields or {}
    match_result = match_result or {}
    recommendation_result = recommendation_result or {}

    max_points = max(1, int(max_evidence_points or 1))
    max_chars = max(1024, int(max_context_chars or 1024))

    selected = _select_relevant_proofs(
        evidence_index,
        confirmed_job_fields,
        match_result,
        max_points=max_points,
    )

    # Always start by shortening long claim texts.
    selected = _shorten_claim_texts(selected, _CLAIM_TEXT_CAP_DEFAULT)

    # Skills / tools / industries / services pulled from match engine
    # (which already filtered by the job's required skills).
    skill_match = match_result.get("skill_match") or {}
    matched_skills = [str(s) for s in (skill_match.get("matched") or [])]
    missing_skills = [str(s) for s in (skill_match.get("missing") or [])]
    industry_match = match_result.get("industry_match") or {}
    matched_industries = [str(s) for s in (industry_match.get("matched") or [])]

    # Tools surfaced inside the selected evidence — small list only.
    tool_set: list[str] = []
    service_set: list[str] = []
    for proof in selected:
        for tool in _proof_attr(proof, "tools", []) or []:
            tool = str(tool).strip()
            if tool and tool not in tool_set:
                tool_set.append(tool)
                if len(tool_set) >= 12:
                    break
        if _proof_attr(proof, "claim_type") in {"service", "deliverable"}:
            text = str(_proof_attr(proof, "claim_text", "") or "").strip()
            if text and text not in service_set:
                service_set.append(text)

    # Selected offer / pricing only if a proof of that type made the cut.
    selected_offer: Optional[str] = None
    for proof in selected:
        if _proof_attr(proof, "claim_type") == "selected_offer":
            selected_offer = _trim_claim_text(
                str(_proof_attr(proof, "claim_text", "") or ""), 200
            )
            break

    pricing: Optional[str] = None
    job_budget = _field(confirmed_job_fields, "budget_or_rate")
    if not is_missing(job_budget):
        for proof in evidence_index or []:
            if _proof_attr(proof, "claim_type") == "pricing":
                pricing = _trim_claim_text(
                    str(_proof_attr(proof, "claim_text", "") or ""), 200
                )
                break

    # Missing-info warnings: critical screenshot fields + recommendation concerns.
    missing_info: list[str] = []
    for key in match_result.get("missing_critical_fields") or []:
        missing_info.append(f"Job field missing from screenshot: {key}")
    if missing_skills:
        missing_info.append(
            "No demonstrated overlap yet for: " + ", ".join(missing_skills[:5])
        )

    context: dict[str, Any] = {
        "confirmed_job_fields": _compact_job_fields(confirmed_job_fields),
        "recommendation": {
            "verdict": recommendation_result.get("verdict"),
            "proposal_angle": recommendation_result.get("best_proposal_angle")
            or recommendation_result.get("proposal_angle")
            or match_result.get("proposal_angle"),
        },
        "matched_skills": matched_skills[:12],
        "matched_industries": matched_industries[:8],
        "tools": tool_set,
        "services": service_set[:6],
        "selected_offer": selected_offer,
        "pricing": pricing,
        "missing_info": missing_info[:6],
        "evidence": _evidence_summaries(selected, cap=_CLAIM_TEXT_CAP_DEFAULT),
    }

    approx_chars = _approx_chars(context)
    # Apply the budget after rendering the actual writer request, including
    # profile and analysis. This object alone is not the request.
    cap = _CLAIM_TEXT_CAP_DEFAULT
    context["__meta__"] = {
        "evidence_points_selected": len(selected),
        "approx_context_chars": approx_chars,
        "claim_text_cap": cap,
        "max_evidence_points": max_points,
        "max_context_chars": max_chars,
    }
    # The compact subset is also returned as Python objects so the
    # verification pass can use the same trimmed list.
    context["__compact_evidence__"] = selected
    return context


def _compact_job_fields(confirmed_job: dict) -> dict[str, str]:
    """Return job fields as a flat dict, dropping unfilled values."""
    out: dict[str, str] = {}
    for key, entry in (confirmed_job or {}).items():
        if isinstance(entry, dict):
            value = str(entry.get("value", "") or "").strip()
        else:
            value = str(entry or "").strip()
        if value and not is_missing(value):
            out[key] = value
    return out


def _evidence_summaries(proofs: list, *, cap: int) -> list[dict[str, Any]]:
    summaries: list[dict[str, Any]] = []
    for proof in proofs:
        summaries.append(
            {
                "evidence_id": _proof_attr(proof, "evidence_id"),
                "claim_type": _proof_attr(proof, "claim_type"),
                "source_type": _proof_attr(proof, "source_type"),
                "claim": _trim_claim_text(
                    str(_proof_attr(proof, "claim_text", "") or ""), cap
                ),
            }
        )
    return summaries


def _approx_chars(context: dict) -> int:
    """Cheap proxy for the eventual prompt size — JSON length of the
    compact context plus a small fixed allowance for prompt scaffolding."""
    serializable = {k: v for k, v in context.items() if not k.startswith("__")}
    try:
        return len(json.dumps(serializable, ensure_ascii=False))
    except (TypeError, ValueError):
        return sum(len(str(v)) for v in serializable.values())


# ---------------------------------------------------------------------------
# LLM call
# ---------------------------------------------------------------------------


def _reply_cut_off(result: Any) -> bool:
    """True when the writer's reply was cut off by the output budget."""
    return getattr(result, "status", None) == llm_client.STATUS_TRUNCATED


def _is_size_error(result: Any) -> bool:
    """True when a smaller request may succeed where this one failed.

    A reply cut off by the output budget is not a size error: it needs more
    room to answer, never less (see :func:`_reply_cut_off`). Nor is an
    account out of credit, which no request size gets past.
    """
    if result is None or _reply_cut_off(result):
        return False
    status = getattr(result, "status", None)
    message = (getattr(result, "error_message", "") or "").lower()
    if any(phrase in message for phrase in llm_client._NO_CREDIT_PHRASES):
        return False
    # A context-window overflow is unambiguously a size problem; a quota /
    # rate-limit (429, tokens-per-minute) is also worth one shrink-retry on
    # the proposal path since a smaller request may slip under a TPM cap.
    if status in (
        getattr(llm_client, "STATUS_CONTEXT_OVERFLOW", "context_overflow"),
        llm_client.STATUS_QUOTA,
    ):
        return True
    return any(
        marker in message
        for marker in (
            "429",
            "rate limit",
            "rate_limit",
            "tokens per minute",
            "context length",
            "context_length",
            "context window",
            "too many tokens",
            "request too large",
        )
    )


def _build_profile_block(canonical_profile: Any, *, compact: bool = False) -> str:
    """Build a compact, proposal-focused profile summary from canonical_profile.

    Includes the Upwork profile optimisation content (title, positioning,
    offer, deliverables, skills, personality, unique element) so the LLM
    can anchor the hook and proof sections in the student's actual profile.

    ``compact`` is for the size retry: fewer and shorter list entries.
    """
    if canonical_profile is None:
        return "{}"
    list_cap, extra_list_cap, item_cap, text_cap = (
        (_COMPACT_PROFILE_LIST_CAP, _COMPACT_PROFILE_LIST_CAP, 100, 150)
        if compact
        else (_PROFILE_LIST_CAP, 6, 200, 300)
    )

    def _val(name: str) -> Any:
        field = getattr(canonical_profile, name, None)
        if field is None:
            if isinstance(canonical_profile, dict):
                entry = canonical_profile.get(name)
                return entry.get("value") if isinstance(entry, dict) else entry
            return None
        return getattr(field, "value", None)

    summary: dict[str, Any] = {}
    for name in (
        "title_or_positioning",   # Upwork title / positioning statement
        "selected_offer",          # the specific offer they lead with
        "target_client",           # who they serve
        "services",                # services they provide
        "deliverables",            # specific deliverables
        "skills",                  # skill list
        "tools",                   # tools they use
        "industries",              # industries they work in
        "strengths",               # their strengths / personality trait
        "work_history",            # past roles / experience
        "achievements",            # quantified achievements
        "pricing",                 # rate / pricing
        "weaknesses_to_account_for",  # things NOT to claim (tools marked basic)
    ):
        value = _val(name)
        if isinstance(value, list):
            # A long list (every weakness a call transcript mentions, every
            # deliverable in a dossier) would crowd out the evidence.
            value = [str(v)[:item_cap] for v in value if v][:list_cap]
        if value:
            summary[name] = value if not isinstance(value, str) else value[:text_cap]

    # How the person wants to be represented, and the working facts a job
    # post often asks about. The prompt says how to use each one; without
    # them here, answers such as a preferred tone or a stated guarantee
    # reached the writer only as loose evidence, if at all. Capped so a
    # long list from a dossier file cannot crowd out the evidence.
    for name in (
        "proposal_preferences",    # tone, how to open, stance on test tasks
        "guarantee",               # promises they actually make
        "preferred_project_types", # availability, hours, project setups
        "timezone",
        "languages",
    ):
        value = _val(name)
        if not value:
            continue
        if isinstance(value, list):
            items = [str(v)[:item_cap] for v in value if v][:extra_list_cap]
            if items:
                summary[name] = items
        else:
            summary[name] = str(value)[:text_cap]
    return json.dumps(summary, ensure_ascii=False, indent=2) if summary else "{}"


_PROFILE_LIST_CAP = 8
_COMPACT_PROFILE_LIST_CAP = 2

# A reply cut off by the output budget is retried with twice the budget, at
# least the first of these and at most the second. Every listed Anthropic
# and Gemini model accepts the cap, and a reply that long is far past a
# proposal's size (OpenAI and Groq are not sent a budget at all).
_TRUNCATION_RETRY_MIN_TOKENS = 4000
_MAX_PROPOSAL_OUTPUT_TOKENS = 8192

# The least output room the writer's reply needs: the proposal, its claims
# and three critique rounds, plus one answer per screening question.
_REPLY_BASE_TOKENS = 2600
_REPLY_TOKENS_PER_SCREENING_ANSWER = 175


def _first_output_budget(settings, confirmed_job: dict) -> int:
    """Output tokens for the first writer call.

    The configured budget, raised to what this job's reply needs. A budget
    saved in Setup before the default went up (2,000) stays in ``.env`` and
    would cut off the reply for a job with several screening questions; a
    cut-off reply is still paid for, so a budget that small saves nothing.
    """
    questions = _field(confirmed_job, "screening_questions")
    count = 0 if is_missing(questions) else sum(
        1 for line in questions.splitlines() if line.strip()
    )
    needed = _REPLY_BASE_TOKENS + _REPLY_TOKENS_PER_SCREENING_ANSWER * count
    return max(
        settings.proposal_max_output_tokens, min(needed, _MAX_PROPOSAL_OUTPUT_TOKENS)
    )


# Groq's free plan allows roughly 8,000 tokens a minute for its default model,
# and the full proposal request is about that size before any answer. On Groq
# the evidence is trimmed to fit, and the extra critique re-ask (a second
# full-size call in the same minute) is skipped.
_GROQ_MAX_EVIDENCE_POINTS = 10
_GROQ_MAX_CONTEXT_CHARS = 6000


_CRITIQUE_CORRECTION = """

CORRECTION — your previous response did not return three completed critique
rounds. Run the THREE-ROUND CRITIQUE in full, then return "critique_rounds" as
a list of exactly three objects, each with a non-empty "findings". The
"proposal" field must be the Round 3 draft. Return the JSON object only.
"""


def _has_three_critique_rounds(payload: Any) -> bool:
    """True when the payload reports three non-empty critique rounds.

    The prompt requires the model to critique its draft three times before
    settling on a final proposal. The rounds are self-reported, so this is a
    presence check, not a proof that the revision actually happened.
    """
    rounds = payload.get("critique_rounds") if isinstance(payload, dict) else None
    if not isinstance(rounds, list) or len(rounds) != 3:
        return False
    return all(
        isinstance(entry, dict) and str(entry.get("findings") or "").strip()
        for entry in rounds
    )


def _cited_ids(value: Any) -> list[Optional[str]]:
    """The evidence ids one writer claim cites: one entry per claim to keep.

    The prompt asks for a single ``evidence_id`` string; a claim citing two
    (``["ev_1", "ev_2"]``) becomes one claim per id. Anything that is not an
    id counts as none, never as the text "None".
    """
    if isinstance(value, str):
        return [value.strip() or None]
    if isinstance(value, (list, tuple)):
        ids = [v.strip() for v in value if isinstance(v, str) and v.strip()]
        return ids or [None]
    return [None]


def _try_llm_generate(
    confirmed_job: dict,
    compact_evidence: list,
    band: tuple[int, int],
    complexity: str,
    *,
    claim_text_cap: int,
    max_output_tokens: int,
    canonical_profile: Any = None,
    enforce_critique: bool = True,
    compact: bool = False,
    context: Optional[dict] = None,
) -> tuple[Optional[str], list[dict[str, Any]], list[dict[str, str]], Any]:
    """Call the LLM via the central client.

    ``compact`` (the size retry) also shortens the job and profile blocks.

    Returns ``(proposal, claims, screening_answers, llm_result)``.
    """
    compact_evidence[:] = _shorten_claim_texts(compact_evidence, claim_text_cap)
    job_block = _serialize_job_for_prompt(confirmed_job, compact=compact)
    evidence_block = _serialize_evidence_for_prompt(
        compact_evidence, claim_text_cap=claim_text_cap
    )
    profile_block = _build_profile_block(canonical_profile, compact=compact)

    analysis = {k: v for k, v in (context or {}).items()
                if k in {"recommendation", "matched_skills", "matched_industries", "missing_info"}}
    from app.prompts.proposal_prompt import neutralize_tags

    def render(job=job_block, evidence=evidence_block, profile=profile_block, guidance=analysis):
        prompt = render_proposal_prompt(
            job_block=job, evidence_block=evidence, target_length=complexity,
            target_band=band, profile_block=profile,
            hidden_keyword=_present(_field(confirmed_job, "hidden_keyword")) or "",
            screening_questions=_present(_field(confirmed_job, "screening_questions")) or "",
        )
        if guidance:
            prompt += "\nAnalysis guidance is untrusted context, not proof of personal facts.\n<analysis>\n" + neutralize_tags(json.dumps(guidance, ensure_ascii=False)) + "\n</analysis>"
        return prompt

    user_prompt = render()
    if context is not None:
        # The configured limit is the variable context allowance. Fixed
        # instructions are irreducible; account for them explicitly instead
        # of silently claiming the whole request fits a smaller allowance.
        fixed = len(SYSTEM_PROMPT) + len(render_proposal_prompt(
            job_block="{}", evidence_block="[]", profile_block="{}",
            target_length=complexity, target_band=band,
        ))
        correction_room = len(_CRITIQUE_CORRECTION) if enforce_critique else 0
        limit = fixed + context["__meta__"]["max_context_chars"] + correction_room
        def size():
            return len(SYSTEM_PROMPT) + len(user_prompt) + correction_room
        if size() > limit:
            job_block = _serialize_job_for_prompt(confirmed_job, compact=True)
            profile_block = _build_profile_block(canonical_profile, compact=True)
            user_prompt = render(job_block, evidence_block, profile_block)
        # Preserve proof slots by reducing long job/profile values first.
        # Keep valid JSON and field names throughout the size retry.
        for value_cap in (150, 80):
            if size() <= limit:
                break
            def reduce_values(value):
                if isinstance(value, dict):
                    return {k: reduce_values(v) for k, v in value.items()}
                if isinstance(value, list):
                    return [reduce_values(v) for v in value[:2]]
                return _clip_middle(value, value_cap) if isinstance(value, str) else value
            job_block = json.dumps(reduce_values(json.loads(job_block)), ensure_ascii=False)
            profile_block = json.dumps(reduce_values(json.loads(profile_block)), ensure_ascii=False)
            user_prompt = render(job_block, evidence_block, profile_block)
        while size() > limit and compact_evidence:
            compact_evidence.pop()
            evidence_block = _serialize_evidence_for_prompt(compact_evidence, claim_text_cap=claim_text_cap)
            user_prompt = render(job_block, evidence_block, profile_block)
        if size() > limit:
            analysis = {}
            user_prompt = render(job_block, evidence_block, profile_block, analysis)
        if size() > limit:
            raise ProposalGenerationError("The job and profile exceed the context allowance. Shorten the inputs or raise the context limit.", status="context_overflow")
        context["__profile_block__"] = profile_block
        context["__meta__"].update(
            evidence_points_selected=len(compact_evidence),
            approx_context_chars=len(SYSTEM_PROMPT) + len(user_prompt),
            fixed_prompt_chars=fixed,
            rendered_request_limit=limit,
        )
    result = llm_client.call_text_llm(
        task_name=TASK_NAME,
        system_prompt=SYSTEM_PROMPT,
        user_prompt=user_prompt,
        expected_json=True,
        max_tokens=max_output_tokens,
    )

    # Enforce the three-round critique. Only re-ask when the call itself
    # succeeded and simply skipped the rounds — a failed or oversized call
    # falls through untouched so the existing size-error retry still owns it.
    if (
        enforce_critique
        and result.success
        and isinstance(result.response_json, dict)
        and not _has_three_critique_rounds(result.response_json)
    ):
        reask = llm_client.call_text_llm(
            task_name=TASK_NAME,
            system_prompt=SYSTEM_PROMPT,
            user_prompt=user_prompt + _CRITIQUE_CORRECTION,
            expected_json=True,
            max_tokens=max_output_tokens,
        )
        if context is not None:
            context["__meta__"]["approx_context_chars"] = (
                len(SYSTEM_PROMPT) + len(user_prompt) + len(_CRITIQUE_CORRECTION)
            )
        # The rounds are self-reported, so the re-ask is best effort: a
        # re-ask that failed, was cut off or came back without a proposal
        # never costs the user the complete draft already in hand.
        if (
            reask.success
            and isinstance(reask.response_json, dict)
            and plain_text(reask.response_json.get("proposal"))
        ):
            result = reask

    if not result.success or not isinstance(result.response_json, dict):
        return None, [], [], result

    payload = result.response_json
    # Valid JSON is not always the shape the prompt asks for: a proposal
    # sent as a list or dict of sections is joined into paragraphs, a claim
    # that cites several ids becomes one claim per id, and screening answers
    # sent as a {question: answer} mapping or as bare strings are kept.
    proposal_text = plain_text(payload.get("proposal"))
    claims_raw = payload.get("factual_claims") or []
    claims: list[dict[str, Any]] = []
    if isinstance(claims_raw, list):
        for item in claims_raw:
            if not isinstance(item, dict):
                continue
            for evidence_id in _cited_ids(item.get("evidence_id")):
                claims.append(
                    {
                        "text": str(item.get("text") or item.get("claim") or "").strip(),
                        "kind": str(item.get("kind") or item.get("claim_type") or "claim"),
                        "evidence_id": evidence_id,
                        "source_file": next((_proof_attr(p, "source_file") for p in compact_evidence if _proof_attr(p, "evidence_id") == evidence_id), None),
                        "claim_type": item.get("claim_type"),
                    }
                )
    screening: list[dict[str, str]] = []
    raw_answers = payload.get("screening_answers") or []
    if isinstance(raw_answers, dict):
        raw_answers = (
            [raw_answers]
            if "answer" in raw_answers
            else [{"question": q, "answer": a} for q, a in raw_answers.items()]
        )
    if isinstance(raw_answers, list):
        for item in raw_answers:
            if isinstance(item, str):
                item = {"answer": item}
            if not isinstance(item, dict):
                continue
            question = str(item.get("question") or "").strip()
            answer = plain_text(item.get("answer"))
            if answer:
                screening.append({"question": question, "answer": answer})
    if not proposal_text:
        return None, claims, screening, result
    return proposal_text, claims, screening, result


def _verification_fields(
    proposal: str,
    compact_evidence: list,
    claims: list[dict[str, Any]],
    confirmed_job: dict,
    settings,
    profile_block: str = "{}",
) -> dict[str, Any]:
    """Fact-check ``proposal`` and return the result fields that report it.

    The verification pass receives ONLY the compact evidence subset that
    produced the draft — never the full evidence index.
    """
    kwargs: dict[str, Any] = {
        "factual_claims": claims,
        "confirmed_job_fields": confirmed_job,
        "settings": settings,
        "profile_block": profile_block,
    }
    try:
        inspect.signature(run_verification).bind(proposal, compact_evidence, **kwargs)
    except TypeError:
        # Backwards-compat: callers (or tests) may monkey-patch
        # run_verification with the older positional-only signature. The
        # signature is checked before the call, so a TypeError raised inside
        # a check is never taken for it and the paid call never runs twice.
        kwargs = {"factual_claims": claims}
    except ValueError:
        pass  # no signature to read: call it with the full arguments
    report = run_verification(proposal, compact_evidence, **kwargs)
    verification_status = getattr(report, "verification_status", "skipped")
    # Grounding gate: the text shown is the checked text — the verifier's
    # cleaned rewrite, or the draft itself when the check could not run
    # ("unverified"). Neither is thrown away: the UI marks a draft the check
    # did not confirm instead of hiding it. The draft only stands in for an
    # empty rewrite when the check flagged nothing.
    verified_text = report.cleaned_proposal or ""
    if (
        not verified_text
        and verification_status != "failed"
        and not (report.removed_claims or report.softened_claims)
    ):
        verified_text = proposal
    # Evidence IDs are useful internally but never belong in the
    # client-facing proposal copy.
    verified_text = strip_evidence_ids(verified_text)
    # Final guard: strip any stray HTML tags the verification pass may have
    # reintroduced, so the stored text (copy + download, not just display)
    # never contains a stray </div> or similar.
    if verified_text:
        verified_text = _strip_forbidden(
            verified_text, hidden_keyword=_field(confirmed_job, "hidden_keyword")
        )
    return {
        "proposal": verified_text,
        "verified_proposal": verified_text,
        "factual_claims": report.surviving_claims,
        "word_count": _word_count(verified_text),
        "removed_claims": report.removed_claims,
        "softened_claims": report.softened_claims,
        "flagged_as_missing": report.flagged_as_missing,
        "supported_claims": getattr(report, "supported_claims", []),
        "partially_supported_claims": getattr(
            report, "partially_supported_claims", []
        ),
        "unsupported_claims": getattr(report, "unsupported_claims", []),
        "verification_status": verification_status,
        "verification_summary": getattr(report, "summary", ""),
        "verification_meta": getattr(report, "meta", {}),
    }


def generate(
    confirmed_job: dict,
    evidence_index: Iterable,
    recommendation: dict,
    *,
    match_data: Optional[dict] = None,
    canonical_profile: Any = None,
    complexity: Optional[str] = None,
    run_verify: bool = True,
    settings=None,
) -> dict[str, Any]:
    """Draft a grounded proposal.

    The default path calls the configured LLM via
    :mod:`app.services.llm_client` with task ``proposal_generation``.
    Only a compact, relevance-filtered subset of the evidence index is
    ever serialized into the prompt — the full dossier, raw transcript
    text, or PDF body is never sent.

    If the first call fails because the request was too large for the
    provider's rate / token / context limits, the call is retried once
    with a smaller request (at most 8 evidence points with shorter
    claims, and a shortened profile and job post). If the reply was cut
    off by the output token budget instead, the same request is retried
    once with a larger budget. No retry lowers the output budget. If the
    retry also fails this raises :class:`ProposalGenerationError` with a
    sanitized message — the UI catches it and shows a clean error
    instead of the raw provider response.

    If the LLM call fails *and* ``ALLOW_LOCAL_PLACEHOLDERS`` is true in
    settings, a deterministic local draft is produced and clearly
    flagged as a placeholder.
    """
    settings = settings or get_settings()
    evidence_list = list(evidence_index or [])
    complexity = complexity or _detect_complexity(confirmed_job)
    band = LENGTH_BANDS.get(complexity, LENGTH_BANDS["standard"])

    matched_skills: list[str] = []
    if match_data:
        matched_skills = list(match_data.get("skill_match", {}).get("matched", []) or [])

    on_groq = getattr(settings, "active_provider", "") == "groq"
    max_points = settings.max_proposal_evidence_points
    max_chars = settings.max_proposal_context_chars
    if on_groq:
        max_points = min(max_points, _GROQ_MAX_EVIDENCE_POINTS)
        max_chars = min(max_chars, _GROQ_MAX_CONTEXT_CHARS)
    primary_context = build_proposal_context(
        confirmed_job,
        evidence_list,
        match_result=match_data,
        recommendation_result=recommendation,
        max_evidence_points=max_points,
        max_context_chars=max_chars,
    )

    used_api = False
    llm_status = "skipped"
    llm_provider = None
    llm_model = None
    llm_error: Optional[str] = None
    sanitized_error: Optional[str] = None

    proposal: Optional[str] = None
    claims: list[dict[str, Any]] = []
    screening_answers: list[dict[str, str]] = []
    llm_result = None
    retry_used = False
    shrunk = False
    chosen_context = primary_context

    if settings.has_api_key:
        compact_meta = primary_context["__meta__"]
        compact_evidence = primary_context["__compact_evidence__"]
        first_budget = _first_output_budget(settings, confirmed_job)
        proposal, claims, screening_answers, llm_result = _try_llm_generate(
            confirmed_job,
            compact_evidence,
            band,
            complexity,
            claim_text_cap=compact_meta["claim_text_cap"],
            max_output_tokens=first_budget,
            canonical_profile=canonical_profile,
            enforce_critique=not on_groq,
            context=primary_context,
        )
        if llm_result is not None:
            llm_status = llm_result.status
            llm_provider = llm_result.provider
            llm_model = llm_result.model
            llm_error = llm_result.error_message
            sanitized_error = llm_client.sanitize_error_message(llm_error)
            used_api = bool(llm_result.used_api and proposal)

        # One retry, chosen by why the first try failed. Neither kind ever
        # lowers the output budget.
        retry_budget = first_budget
        if proposal is None and _reply_cut_off(llm_result):
            # The reply ran out of room before it finished. At temperature 0
            # the same budget cuts it off at the same place, so the same
            # request goes again with more room (if the cap allows more).
            retry_budget = min(
                max(2 * first_budget, _TRUNCATION_RETRY_MIN_TOKENS),
                _MAX_PROPOSAL_OUTPUT_TOKENS,
            )
            retry_used = retry_budget > first_budget
        elif proposal is None and _is_size_error(llm_result):
            # The request was too large: shrink it and retry ONCE. Always
            # smaller than the first try: 8 points / 8,000 chars on the
            # default budget (proportionally less on Groq's smaller one),
            # shorter claims, and a shortened profile and job post.
            retry_used = shrunk = True
            chosen_context = build_proposal_context(
                confirmed_job,
                evidence_list,
                match_result=match_data,
                recommendation_result=recommendation,
                max_evidence_points=min(8, max(4, max_points - 4)),
                max_context_chars=min(8000, int(max_chars * 0.4)),
            )
        if retry_used:
            retry_meta = chosen_context["__meta__"]
            claim_text_cap = retry_meta["claim_text_cap"]
            if shrunk:
                claim_text_cap = min(claim_text_cap, _CLAIM_TEXT_CAP_SMALL)
            proposal, claims, screening_answers, llm_result = _try_llm_generate(
                confirmed_job,
                chosen_context["__compact_evidence__"],
                band,
                complexity,
                claim_text_cap=claim_text_cap,
                max_output_tokens=retry_budget,
                canonical_profile=canonical_profile,
                enforce_critique=not on_groq,
                compact=shrunk,
                context=chosen_context,
            )
            if llm_result is not None:
                llm_status = llm_result.status
                llm_provider = llm_result.provider
                llm_model = llm_result.model
                llm_error = llm_result.error_message
                sanitized_error = llm_client.sanitize_error_message(llm_error)
                used_api = bool(llm_result.used_api and proposal)

    final_meta = chosen_context["__meta__"]
    evidence_points_sent = final_meta["evidence_points_selected"] if used_api else 0
    compact_context_chars = final_meta["approx_context_chars"] if used_api else 0

    # Record proposal-specific metadata onto the most recent
    # proposal_generation entry that the central client wrote. Never
    # log raw prompts, raw dossier text, raw proposals, or API keys.
    if settings.has_api_key:
        llm_client.extend_last_entry(
            TASK_NAME,
            {
                "evidence_points_sent": final_meta["evidence_points_selected"],
                "compact_context_chars": final_meta["approx_context_chars"],
                "retry_used": retry_used,
                "sanitized_error_message": sanitized_error,
            },
        )

    if proposal is None:
        if not settings.allow_local_placeholders:
            # Failure path — surface a clean message to the UI. The
            # provider error (already sanitized of organization IDs) is
            # attached separately for debug surfaces.
            if not settings.has_api_key:
                message = (
                    "Proposal generation failed because no LLM API key is "
                    "configured. Set ANTHROPIC_API_KEY or OPENAI_API_KEY in "
                    "your .env, or enable ALLOW_LOCAL_PLACEHOLDERS=true to "
                    "run the deterministic local draft for development."
                )
            elif _reply_cut_off(llm_result):
                message = TRUNCATION_FAILURE_USER_MESSAGE
            elif shrunk:
                message = SIZE_FAILURE_USER_MESSAGE
            else:
                message = "Proposal generation failed because the LLM API call failed."

            raise ProposalGenerationError(
                message,
                status=llm_status,
                llm_result=llm_result,
                sanitized_error=sanitized_error,
                meta={
                    "task_name": TASK_NAME,
                    "provider": llm_provider,
                    "model": llm_model,
                    "evidence_points_sent": final_meta["evidence_points_selected"],
                    "compact_context_chars": final_meta["approx_context_chars"],
                    "retry_used": retry_used,
                },
            )

        # Local placeholder path — explicitly opted in via env flag.
        proposal, claims = _local_generate(
            confirmed_job, evidence_list, matched_skills, band
        )
        llm_client.record_local_use(
            TASK_NAME,
            note="ALLOW_LOCAL_PLACEHOLDERS=true; deterministic draft used.",
        )

    proposal = _strip_forbidden(
        proposal, hidden_keyword=_field(confirmed_job, "hidden_keyword")
    )
    proposal = _trim_to_band(proposal, band)

    result: dict[str, Any] = {
        "draft_proposal": proposal,
        "proposal": proposal,
        "factual_claims": claims,
        "screening_answers": screening_answers,
        "target_length": band,
        "complexity": complexity,
        "word_count": _word_count(proposal),
        "__meta__": {
            "task_name": TASK_NAME,
            "used_api": used_api,
            "status": llm_status if used_api else "local_placeholder",
            "provider": llm_provider,
            "model": llm_model,
            "error_message": sanitized_error if not used_api else None,
            "sanitized_error_message": sanitized_error,
            "evidence_points_sent": evidence_points_sent,
            "compact_context_chars": compact_context_chars,
            "retry_used": retry_used,
        },
    }

    # The verification pass receives ONLY the compact evidence subset —
    # never the full evidence index.
    compact_for_verify = chosen_context.get("__compact_evidence__") or []

    if run_verify:
        # Kept so the fact-check can run again on this same draft
        # (recheck_proposal) when it could not run the first time.
        result["__verify_inputs__"] = {
            "evidence": compact_for_verify,
            "claims": claims,
            "profile_block": chosen_context.get("__profile_block__", _build_profile_block(canonical_profile, compact=shrunk)),
            "screening_answers": screening_answers,
        }
        result.update(
            _verification_fields(
                proposal, compact_for_verify, claims, confirmed_job, settings,
                result["__verify_inputs__"]["profile_block"],
            )
        )
    else:
        # Always strip evidence IDs from the visible proposal even when
        # verification is skipped — they must never reach the client copy.
        stripped = strip_evidence_ids(proposal)
        result.update(
            {
                "proposal": stripped,
                "verified_proposal": stripped,
                "word_count": _word_count(stripped),
                "removed_claims": [],
                "softened_claims": [],
                "flagged_as_missing": [],
                "supported_claims": [],
                "partially_supported_claims": [],
                "unsupported_claims": [],
                "verification_status": "skipped",
                "verification_summary": "Verification pass was not run.",
                "verification_meta": {
                    "task_name": "verification_pass",
                    "used_api": False,
                    "status": "skipped",
                    "error_message": "Verification pass was not run.",
                    "claims_checked": 0,
                    "unsupported_claims_count": 0,
                },
            }
        )

    if screening_answers:
        if run_verify:
            result["screening_answers"] = _check_screening(
                screening_answers, compact_for_verify, confirmed_job, settings,
                result["__verify_inputs__"]["profile_block"],
            )
        else:
            result["screening_answers"] = [
                {**qa, "answer": _strip_forbidden(strip_evidence_ids(qa["answer"])), "verification_status": "skipped"}
                for qa in screening_answers
            ]
    return result


def _check_screening(answers, evidence, confirmed_job, settings, profile_block):
    checked = []
    for qa in answers:
        draft = _strip_forbidden(strip_evidence_ids(plain_text(qa.get("answer"))))
        fields = _verification_fields(draft, evidence, [], confirmed_job, settings, profile_block)
        status = fields["verification_status"]
        answer = _strip_forbidden(strip_evidence_ids(fields["proposal"]))
        checked.append({"question": qa.get("question", ""), "answer": answer if status != "failed" else "",
                        "draft_answer": draft, "verification_status": status})
    return checked


def recheck_proposal(
    proposal_result: dict, confirmed_job: dict, *, settings=None
) -> dict[str, Any]:
    """Run the fact-check again on a draft :func:`generate` already wrote.

    Only the verification pass runs, on the same draft and the same
    evidence subset, so the writer is not paid for again. Used when the
    first check could not run. Returns a new result dict; the one passed
    in is left as it was.
    """
    settings = settings or get_settings()
    result = dict(proposal_result or {})
    inputs = result.get("__verify_inputs__") or {}
    draft = result.get("draft_proposal") or ""
    if draft:
        result.update(
            _verification_fields(
                draft,
                list(inputs.get("evidence") or []),
                list(inputs.get("claims") or []),
                confirmed_job,
                settings,
                inputs.get("profile_block", "{}"),
            )
        )
    if inputs.get("screening_answers"):
        result["screening_answers"] = _check_screening(
            inputs["screening_answers"], list(inputs.get("evidence") or []),
            confirmed_job, settings, inputs.get("profile_block", "{}"),
        )
    return result