app/services/generation.py

"""Generation service: orchestrates notes expansion → retrieval → style analysis → adapt → cache → persist.

Supports three AI modes:

* **generate** — Full RAG pipeline: expand raw notes → retrieve evidence →
  personalise to the user's writing style → generate polished prose.
* **proofread** — Grammar, clarity, and style review of an existing section.
* **enhance** — Technical depth expansion using broader retrieved evidence.

This module is called as a ``BackgroundTask`` from the generate endpoint and
manages its own database session.
"""

import asyncio
import json
import logging
import re
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.agentic.models import StructuredReport
from app.cache import section_cache, style_cache
from app.config import settings
from app.db.database import get_session_factory
from app.db.models import Report, ReportSection, ReportStatus
from app.llm import generation_facade as gen_llm
from app.generator.notes_expander import expand_notes, expand_notes_async
from app.generator.postprocess import (
    async_enforce_verify,
    enforce_verify,
    strip_scaffold_subheadings,
    verbatim_overlap_ratio,
)
from app.generator.prompts import verbatim_ratio_target
from app.generator.section_scope import is_introduction_section, section_scope_block
from app.generator.style_analyzer import analyze_writing_style, analyze_writing_style_async
from app.models.schemas import (
    AILevel,
    GenerationMode,
    Provenance,
    RerankedResult,
    SearchResult,
    WritingStyleProfile,
    ai_level_to_percent,
    ai_percent_to_level,
)
from app.retrieval.reranker import rerank
from app.retrieval.survey_filter import filter_search_results_by_survey_level
from app.services.standard_paragraphs import get_standard_paragraph_for_section
from app.retrieval.retriever import (
    reorder_by_chunk_role,
    retrieve,
    retrieve_async,
    retrieve_document_level_context,
    retrieve_document_level_context_async,
    retrieve_for_report,
    retrieve_for_report_async,
    retrieve_for_report_unified,
    retrieve_scoped_async,
    retrieve_unified,
    use_async_retrieval_path,
)
from app.templates.registry import ALL_VALID_SECTION_CODES
from app.retrieval.vector_search import async_vs_search
from app.services.runtime_rag_index import runtime_section_vector_doc_id
from app.services.ai_transparency import compute_ai_transparency
from app.services.provenance_enrichment import (
    attach_snippet_metadata,
    fetch_doc_filenames,
)
from app.services.photo_policy import PhotoPolicy, classify_section_photo_policy_async
from app.services.photo_vision import get_photo_observations_for_section
from app.templates.registry import get_survey_pack, get_template
from app.vectorstore.factory import get_vectorstore

logger = logging.getLogger(__name__)


def _normalise_interference_level(raw: str | None) -> str | None:
    if not raw:
        return None
    t = str(raw).strip().lower()
    if t == "minimal":
        t = "minimum"
    return t if t in ("minimum", "medium", "maximum") else None


def _interference_from_cache_entry(cached: dict[str, Any]) -> str | None:
    raw = cached.get("interference_level")
    if isinstance(raw, str):
        n = _normalise_interference_level(raw)
        if n:
            return n
    leg = cached.get("ai_involvement_tier")
    if isinstance(leg, str):
        return _normalise_interference_level(leg)
    return None


_MISSING_FACT_SENTENCES: tuple[str, str] = (
    "Information not provided in source document.",
    "We were unable to verify this during inspection.",
)


def _clean_and_clamp_bullets(bullets: list[str], *, max_items: int | None = None) -> list[str]:
    """Normalise messy notes (split dense lines, de-dupe, clamp)."""
    from app.generator.note_bullets import clean_and_clamp_bullets

    cap = max_items if max_items is not None else int(settings.max_bullets_per_section)
    return clean_and_clamp_bullets(bullets, max_items=cap)


def _agentic_inspector_meta(rep: StructuredReport) -> dict[str, Any] | None:
    """Subset of the inspector ``StructuredReport`` safe to JSON-serialize into section meta."""
    out: dict[str, Any] = {}
    if rep.extraction_audit:
        out["extraction_audit"] = rep.extraction_audit
    if rep.section_plan:
        out["section_plan"] = rep.section_plan
    if rep.condition_rating_summary:
        out["condition_rating_summary"] = rep.condition_rating_summary
    if rep.tool_trace:
        out["tool_trace"] = [dict(x) for x in list(rep.tool_trace)[-48:]]
    return out or None

_POSTCODE_RE = re.compile(r"\b([A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2})\b", re.IGNORECASE)
_ADDRESS_LINE_RE = re.compile(
    r"\b(\d{1,4}\s+[A-Za-z][A-Za-z'\-]*(?:\s+[A-Za-z][A-Za-z'\-]*){0,6}\s+"
    r"(Road|Rd|Street|St|Avenue|Ave|Lane|Ln|Drive|Dr|Crescent|Close|Place|Way|Gardens|Gdns|Court|Ct|Terrace|Terr))\b",
    re.IGNORECASE,
)

_MONEY_RE = re.compile(r"(£\s*\d[\d,]*(?:\.\d+)?|\b\d[\d,]*(?:\.\d+)?\s*(?:gbp|pounds)\b)", re.IGNORECASE)
_DATE_RE = re.compile(
    r"\b(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{1,2}\s+(?:jan|feb|mar|apr|may|jun|jul|aug|sep|sept|oct|nov|dec)[a-z]*\s+\d{2,4})\b",
    re.IGNORECASE,
)
_LONG_NUMBER_RE = re.compile(r"\b\d{5,}\b")

def _wrap_structured_skeleton(
    *,
    survey_level: int,
    code: str,
    title: str | None,
    base_skeleton: str,
    has_condition_rating: bool,
) -> str:
    """Standardised sub-template blocks for Levels 1–3 section drafting."""
    t = (title or "").strip()
    heading = f"{code} — {t}" if t else code
    lvl = 3
    try:
        lvl = int(survey_level or 3)
    except Exception:  # noqa: BLE001
        lvl = 3
    lvl = max(1, min(3, lvl))

    rating_note = "Do not invent a condition rating."
    if has_condition_rating:
        rating_note = "Include an explicit Condition Rating (1/2/3/NI) only if it appears in the RAW NOTES."

    if lvl <= 1:
        exec_hint = "[1–3 sentences describing observed condition and limitations (no advice).]"
        rec_hint = "[This Level 1 report does not provide recommendations. Record limitations only.]"
    elif lvl == 2:
        exec_hint = "[1–3 sentences summarising key findings and main next steps.]"
        rec_hint = "[practical next steps / further checks / quotations, proportionate for Level 2, only where supported.]"
    else:
        exec_hint = "[1–3 sentences capturing the main condition message for this element.]"
        rec_hint = "[next steps / further investigations / repair priorities appropriate for Level 3, but only where supported.]"

    focus = (base_skeleton or "").strip()
    if len(focus) > 550:
        focus = focus[:550].rstrip() + "…"

    scope = section_scope_block(code, title)

    # Section D ("About the property") is an INTRODUCTION, not an element report.
    # The generic Executive Summary / Condition / Defects scaffold is exactly
    # what makes the model dump roof/damp/floor detail here and then repeat it
    # in E–K. Give D a tight introduction-only contract instead.
    if is_introduction_section(code):
        return (
            f"{heading}\n\n"
            f"{scope}\n\n"
            "Write 2\u20133 sentences ONLY. State the property type and tenure, "
            "the address and date of inspection if present in the RAW NOTES, "
            "and one sentence on overall character. If the property is listed "
            "or in a conservation area, state that prominently (it constrains "
            "permitted works) without listing detail. If alterations have been "
            "made, note their approval/date is unknown where applicable and "
            "that the legal sections contain the detail \u2014 do not repeat it "
            "here. Do NOT describe any defect, repair, roof, damp, wall, floor "
            "or service condition. "
            'End with exactly: "Full details of each element are provided in '
            'sections E through K of this report."\n\n'
            "Section focus (template guidance)\n"
            f"{focus}"
        )

    return (
        f"{heading}\n\n"
        f"{scope}\n\n"
        "Executive Summary\n"
        f"{exec_hint}\n\n"
        "Property Description\n"
        "[brief factual description relevant to this element, grounded in RAW NOTES]\n\n"
        "Condition Assessment\n"
        f"[observations + condition; {rating_note}]\n\n"
        "Defects and Risks\n"
        "[key defects, mechanisms (only if supported), implications/risks]\n\n"
        "Recommendations\n"
        f"{rec_hint}\n\n"
        "Section focus (template guidance)\n"
        f"{focus}"
    )


def _redact_reference_snippet(text: str) -> str:
    """Redact property-specific facts from RAG snippets used as guidance.

    In notes-only generation mode, retrieved content must never contribute
    factual/property-specific claims (addresses, postcodes, dates, prices,
    long identifiers). We still allow generic phrasing/structure to influence
    the draft, but redact the common fact-shaped patterns.
    """
    t = (text or "").strip()
    if not t:
        return ""
    t = _POSTCODE_RE.sub("[REDACTED_POSTCODE]", t)
    t = _ADDRESS_LINE_RE.sub("[REDACTED_ADDRESS]", t)
    t = _MONEY_RE.sub("[REDACTED_MONEY]", t)
    t = _DATE_RE.sub("[REDACTED_DATE]", t)
    t = _LONG_NUMBER_RE.sub("[REDACTED_NUMBER]", t)
    return t


def _redact_reference_snippets(snips: list[str]) -> list[str]:
    out: list[str] = []
    for s in snips or []:
        rs = _redact_reference_snippet(s)
        if rs:
            out.append(rs)
    return out


def _deterministic_assembly_stitch(
    *,
    skeleton: str,
    bullets: list[str],
    paragraph_snippets: list[str] | None,
    document_snippets: list[str] | None,
    hierarchy_section_snippets: list[str] | None,
    survey_level: int | None,
    redact_references: bool = True,
) -> str:
    """Pure-template assembly: splice retrieved standard wording verbatim with
    site notes from bullets, no LLM involvement.

    Activated when ``ai_percent == 0`` so the output is mathematically
    guaranteed to be 0% AI-generated wording: every clause either came from a
    retrieved standard passage (firm's approved boilerplate, with
    property-specific tokens redacted) or from the inspector's raw notes.

    Returns an empty string when there is no boilerplate to splice — the
    caller then falls back to the LLM path (which uses the hardened assembly
    prompt).
    """
    from app.generator.prompts import _word_target_for_involvement

    para = [s for s in (paragraph_snippets or []) if isinstance(s, str) and s.strip()]
    sec = [s for s in (hierarchy_section_snippets or []) if isinstance(s, str) and s.strip()]
    doc = [s for s in (document_snippets or []) if isinstance(s, str) and s.strip()]

    if redact_references:
        para = [s for s in (_redact_reference_snippet(s) for s in para) if s and s.strip()]
        sec = [s for s in (_redact_reference_snippet(s) for s in sec) if s and s.strip()]
        doc = [s for s in (_redact_reference_snippet(s) for s in doc) if s and s.strip()]

    ordered = para + sec + doc
    if not ordered:
        return ""

    _, max_words = _word_target_for_involvement(survey_level, 0)
    notes_budget = max(40, min(80, max_words // 4))
    body_budget = max(60, max_words - notes_budget)

    parts: list[str] = []
    used = 0
    seen: set[str] = set()
    for snip in ordered:
        snip = snip.strip()
        if not snip:
            continue
        key = snip[:200].lower()
        if key in seen:
            continue
        seen.add(key)
        words = snip.split()
        if used + len(words) > body_budget:
            remaining = body_budget - used
            if remaining < 25:
                break
            partial = " ".join(words[:remaining]).strip()
            for terminator in (". ", "! ", "? "):
                idx = partial.rfind(terminator)
                if idx > 60:
                    partial = partial[: idx + 1].strip()
                    break
            if partial:
                parts.append(partial)
                used = body_budget
            break
        parts.append(snip)
        used += len(words)

    body = "\n\n".join(p for p in parts if p)

    bullet_clauses: list[str] = []
    for b in bullets or []:
        t = str(b or "").strip()
        if not t:
            continue
        if not t.endswith((".", "!", "?")):
            t = t + "."
        bullet_clauses.append(t)
    if bullet_clauses:
        notes_text = " ".join(bullet_clauses)
        nwords = notes_text.split()
        if len(nwords) > notes_budget:
            notes_text = " ".join(nwords[:notes_budget]).rstrip(",;:") + "…"
        body = (body + "\n\nSite-specific observations from this inspection: " + notes_text).strip()

    return body
# Surveyor / firm extraction. Catches "Behrang Dizaji MRICS" as a personal name
# (multi-word capitalised phrase ending in MRICS/AssocRICS/FRICS), and
# "Arnold & Baldwin Chartered Surveyors" or similar firm names referenced
# either by the leading "Company name" header in the RICS template or by the
# trailing "Chartered Surveyors" suffix.
_SURVEYOR_NAME_RE = re.compile(
    r"\b([A-Z][A-Za-z'\-]+(?:\s+[A-Z][A-Za-z'\-]+){1,4})\s+(MRICS|FRICS|AssocRICS)\b"
)
_FIRM_AFTER_LABEL_RE = re.compile(
    r"(?im)^\s*Company\s+name\s*:?\s*\n?\s*([A-Za-z0-9&'\-\.\s,]{4,80})\s*$"
)
_FIRM_INLINE_RE = re.compile(
    r"\b([A-Z][A-Za-z'&\-]+(?:\s+[A-Z][A-Za-z'&\-]+){0,4}\s+Chartered\s+Surveyors)\b"
)


def resolve_property_type(text: str) -> dict[str, str]:
    """Reconcile contradictory property-type signals into one coherent string.

    The notes frequently contain a *current legal tenure* statement ("it's a
    house, advise the solicitor on a freehold house") AND a *past alteration*
    ("converted to flats, date unknown"). The LLM, generating each section in
    isolation, lists both as simultaneous truths and writes the incoherent
    "is a house ... it has been converted to flats". This resolves the signals
    deterministically BEFORE generation so every section receives one settled
    description.

    Returns a dict with optional keys: ``property_type`` and ``tenure_note``.
    """
    low = (text or "").lower()
    out: dict[str, str] = {}

    has_house = bool(re.search(r"\bhouse\b", low))
    has_bungalow = bool(re.search(r"\bbungalow\b", low))
    has_maisonette = bool(re.search(r"\bmaisonette\b", low))
    converted = bool(
        re.search(r"convert(?:ed|sion)\s+(?:in)?to\s+flats?\b", low)
        or re.search(r"\bconverted\b.*\bflats?\b", low)
    )
    has_flat = bool(re.search(r"\bflats?\b", low))

    if has_house and converted:
        out["property_type"] = "house converted to flats"
        out["tenure_note"] = (
            "legal tenure should be verified \u2014 the notes reference a "
            "freehold house but also a conversion to flats; the solicitor "
            "should confirm the current tenure and approvals"
        )
    elif has_bungalow:
        out["property_type"] = "bungalow"
    elif has_maisonette:
        out["property_type"] = "maisonette"
    elif has_house:
        out["property_type"] = "house"
    elif has_flat:
        out["property_type"] = "flat"
    return out


def _extract_property_identity(source_lines: list[str]) -> dict[str, str]:
    """Best-effort extraction of pinned identity facts from trusted notes/bullets.

    Looks for: full address (street + postcode kept together when both
    appear on adjacent lines), postcode, property type, occupancy, surveyor
    name (with MRICS/FRICS/AssocRICS suffix), and firm name (Chartered
    Surveyors). The richer pin set lets the inspector-loop system prompt
    instruct the LLM to NEVER invent surveyor or firm details, fixing the
    "located at Information not provided in source document" pattern that
    RAGAS faithfulness checks surfaced even when the address WAS in the
    retrieved evidence.
    """
    text = "\n".join(source_lines or [])
    out: dict[str, str] = {}

    # Address: try "Address:" first, else first street-like line.
    m = re.search(r"(?im)^\s*(?:address|property address)\s*:\s*(.+?)\s*$", text)
    if m:
        out["address"] = m.group(1).strip()
    else:
        m2 = _ADDRESS_LINE_RE.search(text)
        if m2:
            out["address"] = m2.group(1).strip()

    # Postcode: independent of address — the source PDF often puts the
    # street on one line and the postcode on the next, so we capture both
    # and let _identity_facts_block emit them together. Previously the
    # postcode was *only* recorded as a fallback when no street match was
    # found, which dropped SW4 8QE on a hit like "37 Elms Crescent" and
    # let the LLM omit "London SW4 8QE" from property_description.
    pc = _POSTCODE_RE.search(text)
    if pc:
        out["postcode"] = re.sub(r"\s+", " ", pc.group(1)).strip().upper()

    low = text.lower()
    # Property type. Reconciliation runs FIRST: if the notes contain a
    # house/converted-to-flats contradiction it produces one coherent string
    # plus a tenure-verification note, instead of the LLM listing both facts
    # as simultaneous truths.
    resolved = resolve_property_type(text)
    if resolved.get("property_type"):
        out["property_type"] = resolved["property_type"]
        if resolved.get("tenure_note"):
            out["tenure_note"] = resolved["tenure_note"]
    elif "single-family" in low or "single family" in low or "single-family dwelling" in low:
        out["property_type"] = "single-family house"
    elif "townhouse" in low:
        out["property_type"] = "house (townhouse)"
    elif "block of flats" in low or "communal" in low or "self-contained flats" in low:
        out["property_type"] = "flats / multi-occupancy"
    elif "flat" in low and "house" not in low:
        out["property_type"] = "flat"
    elif "house" in low:
        out["property_type"] = "house"

    # Occupancy
    if "vacant" in low or "unoccupied" in low:
        out["occupancy"] = "vacant"
    if "unfurnished" in low:
        out["occupancy"] = (out.get("occupancy", "") + (", " if out.get("occupancy") else "") + "unfurnished").strip()
    if "occupied" in low and "unoccupied" not in low:
        out["occupancy"] = "occupied"

    # Surveyor name (RICS-suffix-anchored). Take the FIRST match — additional
    # suffixed names tend to be cross-references in the KB, not the surveyor
    # of the report under analysis.
    sm = _SURVEYOR_NAME_RE.search(text)
    if sm:
        out["surveyor_name"] = f"{sm.group(1).strip()} {sm.group(2).strip()}"

    # Firm: prefer an explicit "Company name: X" labelled line, fall back to
    # an inline phrase ending in "Chartered Surveyors". Strip any trailing
    # label text that bled in from a multi-line capture.
    fm = _FIRM_AFTER_LABEL_RE.search(text)
    if fm:
        firm = fm.group(1).strip().rstrip(",")
        # The captured value can include the address that follows on the
        # next line in some templates; trim aggressively to the first newline.
        firm = firm.splitlines()[0].strip()
        if 4 <= len(firm) <= 80:
            out["firm"] = firm
    if "firm" not in out:
        fim = _FIRM_INLINE_RE.search(text)
        if fim:
            out["firm"] = fim.group(1).strip()

    return out


def _identity_facts_block(identity: dict[str, str]) -> str:
    """Format pinned identity facts for the prompt.

    The output is hand-tuned so the LLM treats each line as a hard
    constraint. Postcode is now combined with address into a single pin
    (when both are present) — the LLM is more reliable when the entire
    address is one canonical fact than when postcode is a separate line
    that could be elided independently.
    """
    if not identity:
        return "(Not provided.)"
    parts: list[str] = []
    addr = identity.get("address") or ""
    pc = identity.get("postcode") or ""
    if addr and pc and pc not in addr.upper():
        parts.append(
            f"- Address: {addr}, {pc} (use exactly this combined string; "
            f"do not introduce any other address or postcode, and do not "
            f"write 'Information not provided in source document' for the "
            f"property address — it IS provided right here)"
        )
    elif addr:
        parts.append(
            f"- Address: {addr} (use exactly this; do not introduce any "
            f"other address, and do not write 'Information not provided in "
            f"source document' for the property address — it IS provided "
            f"right here)"
        )
    elif pc:
        parts.append(
            f"- Postcode: {pc} (use exactly this; do not invent postcodes)"
        )
    if identity.get("property_type"):
        ptype = identity["property_type"]
        if "convert" in ptype.lower():
            parts.append(
                f"- Property type: {ptype} (use exactly this RESOLVED "
                f"description; never write 'is a house' and 'converted to "
                f"flats' as two separate present-tense facts in the same "
                f"sentence \u2014 they describe tenure vs a past alteration)"
            )
        else:
            parts.append(
                f"- Property type: {ptype} (do not describe as flats/communal unless explicitly stated)"
            )
    if identity.get("tenure_note"):
        parts.append(f"- Tenure: {identity['tenure_note']}")
    if identity.get("occupancy"):
        parts.append(f"- Occupancy: {identity['occupancy']} (do not claim access restrictions that contradict this)")
    if identity.get("surveyor_name"):
        parts.append(
            f"- Surveyor: {identity['surveyor_name']} (this is THE surveyor "
            f"of record — never invent a different name; never replace it "
            f"with the missing-info phrase)"
        )
    if identity.get("firm"):
        parts.append(
            f"- Firm: {identity['firm']} (this is THE firm of record — "
            f"never invent a different company name)"
        )
    parts.append(
        "- Zero-hallucination rule: do not invent identity facts. If a detail is missing from both notes and "
        "evidence, omit the unsupported claim rather than writing placeholder sentences."
    )
    parts.append(
        "- Shared context rule: you already know the facts above. Do NOT "
        "re-introduce property type, legal status (listed building / "
        "conservation area) or tenure in every section. Reference them only "
        "when directly relevant to the element this section describes; they "
        "belong primarily in Section D and the legal (I) sections."
    )
    return "\n".join(parts)


def _detect_identity_contradictions(text: str, identity: dict[str, str]) -> list[str]:
    """Detect high-severity identity contradictions (address/postcode/type/occupancy) in generated text."""
    issues: list[str] = []
    t = text or ""
    low = t.lower()

    expected_addr = (identity.get("address") or "").strip()
    if expected_addr:
        # Any other street-like address line is a hard fail.
        for m in _ADDRESS_LINE_RE.finditer(t):
            found = m.group(1).strip()
            if expected_addr.lower() not in found.lower():
                issues.append(f"Mentions different address '{found}' (expected '{expected_addr}').")
        # Any postcode that isn't the expected one is also a hard fail.
        expected_pc = re.sub(r"\s+", " ", identity.get("postcode", "")).strip().upper()
        for m in _POSTCODE_RE.finditer(t):
            found_pc = re.sub(r"\s+", " ", m.group(1)).strip().upper()
            if expected_pc and found_pc != expected_pc:
                issues.append(f"Mentions different postcode '{found_pc}' (expected '{expected_pc}').")
    else:
        # If we don't know the address, still treat multi-address output as suspicious.
        addrs = {m.group(1).strip() for m in _ADDRESS_LINE_RE.finditer(t)}
        if len(addrs) >= 2:
            issues.append("Mentions multiple different addresses.")

    pt = (identity.get("property_type") or "").lower()
    if pt and "house" in pt and "flats" not in pt:
        if any(x in low for x in ("block of flats", "self-contained flats", "communal area", "communal", "tenants")):
            issues.append("Describes the property as flats/communal/tenanted despite notes indicating a house.")

    occ = (identity.get("occupancy") or "").lower()
    if "vacant" in occ or "unfurnished" in occ:
        if any(x in low for x in ("furniture", "floor coverings", "immovable furniture", "fixed units limited the inspection")):
            issues.append("Claims contents restricted inspection despite notes indicating vacant/unfurnished.")

    return issues


def _tier_validation_issues(text: str, survey_level: int | None) -> list[str]:
    """Enforce survey-level behavioural differences (lightweight heuristics).

    This is *not* a semantic truth checker (that's handled by `enforce_verify` + RAG).
    It is a behavioural guardrail so:
    - Level 1 reads as condition recording, not advice.
    - Level 3 contains diagnostic layering (cause → implications → options) when discussing defects.
    """
    t = (text or "").strip()
    if not t:
        return ["Empty output"]
    if t in _MISSING_FACT_SENTENCES:
        return []

    try:
        lvl = int(survey_level or 3)
    except Exception:  # noqa: BLE001
        lvl = 3

    low = t.lower()
    issues: list[str] = []

    if lvl <= 1:
        forbidden = (
            "we recommend",
            "you should",
            "should be repaired",
            "should be replaced",
            "repair",
            "replace",
            "recommended",
            "advise",
            "recommendation",
        )
        if any(p in low for p in forbidden):
            issues.append("Level 1 contains repair/advice language (not permitted).")
        return issues

    if lvl == 2:
        # Level 2 permits practical advice; we don't enforce diagnostic layering.
        return []

    # Level 3 diagnostic layering (best-effort):
    # require markers for cause + implication + action when a defect is being discussed.
    cause_markers = ("likely", "may be due", "due to", "possibly", "as a result of")
    implication_markers = ("may lead to", "could lead to", "may result in", "could result in", "risk", "if left")
    action_markers = ("further investigation", "recommend", "should be", "consider", "obtain quotations", "specialist")

    has_cause = any(m in low for m in cause_markers)
    has_implication = any(m in low for m in implication_markers)
    has_action = any(m in low for m in action_markers)

    mentions_defect = any(
        m in low
        for m in (
            "condition rating",
            "defect",
            "damp",
            "crack",
            "leak",
            "decay",
            "rot",
            "movement",
        )
    )

    if mentions_defect and not (has_cause and has_implication and has_action):
        missing: list[str] = []
        if not has_cause:
            missing.append("cause/mechanism")
        if not has_implication:
            missing.append("implication/risk")
        if not has_action:
            missing.append("options/next steps")
        issues.append("Level 3 missing diagnostic layer(s): " + ", ".join(missing))

    return issues


def _parse_validator_result(raw: str) -> tuple[bool, str]:
    """Parse OpenAI validator output into (pass_bool, detail_str)."""
    s = (raw or "").strip()
    if not s:
        return False, "Empty validator output"
    up = s.upper()
    if up.startswith("PASS"):
        return True, s
    if up.startswith("FAIL"):
        return False, s
    # Be strict: unknown format = fail (so we don't silently accept).
    return False, "FAIL: Validator returned unexpected format"


# ── Style profile helper ───────────────────────────────────────────────────────

async def _get_or_build_style_profile(tenant_id: str) -> WritingStyleProfile:
    """Return the style profile for ``tenant_id`` using a four-tier fallback.

    Priority order:
    1. User's cached profile (built from their style corpus or own docs).
    2. Style-corpus build: the user's PAST completed reports indexed under
       ``document_purpose=style_corpus``. This is the "private personalised
       AI" path — the tenant's own voice, scrubbed of PII, learned from
       their finished work rather than from whatever happens to be on
       their desk for the current job.
    3. Fresh analysis of the user's currently indexed report-source docs
       (legacy behaviour — useful when the tenant has uploads but no
       dedicated style corpus yet).
    4. Reference profile built from the KB corpus (Behrang's reports) —
       used only when neither corpus nor uploads exist yet.

    The cache is invalidated on every new upload (including style-corpus
    uploads), so style always reflects the user's latest library.

    Args:
        tenant_id: The tenant whose style to analyse.

    Returns:
        :class:`~app.models.schemas.WritingStyleProfile`.
    """
    # Tier 1 — cache hit
    cached = style_cache.get(tenant_id)
    if cached:
        logger.debug("Style profile cache hit for tenant=%s", tenant_id)
        return cached

    # Tier 2 — style corpus (PAST reports, PII-stripped, purpose=style_corpus)
    from app.services.style_corpus import build_style_corpus_profile

    corpus_profile = await build_style_corpus_profile(tenant_id)
    if corpus_profile is not None:
        logger.info(
            "Style profile built from style_corpus tenant=%s examples=%d",
            tenant_id,
            len(corpus_profile.example_paragraphs),
        )
        style_cache.set(tenant_id, corpus_profile)
        return corpus_profile

    # Tier 3 — analyse the user's existing report-source uploads (legacy path)
    logger.info("Analysing writing style for tenant=%s", tenant_id)
    sample_results = await retrieve_unified(
        query="property survey condition description", tenant_id=tenant_id, k=6
    )
    sample_texts = [r.text for r in sample_results]

    if sample_texts:
        key = (settings.openai_api_key or "").strip()
        if key:
            profile = await analyze_writing_style_async(
                sample_texts=sample_texts,
                openai_api_key=key,
                chat_model=settings.chat_model,
            )
        else:
            profile = await asyncio.to_thread(
                analyze_writing_style,
                sample_texts=sample_texts,
                openai_api_key="",
                chat_model=settings.chat_model,
            )
        style_cache.set(tenant_id, profile)
        return profile

    # Tier 4 — no user documents yet; fall back to reference (KB) style profile
    kb_tenant = settings.knowledge_base_tenant_id
    kb_profile = style_cache.get(kb_tenant)
    if kb_profile:
        logger.info("No documents for tenant=%s — using reference style profile from KB", tenant_id)
        return kb_profile

    # Final fallback: mock profile (no KB either)
    logger.info("No user or KB documents for tenant=%s — using mock style profile", tenant_id)
    from app.generator.style_analyzer import _MOCK_PROFILE

    return _MOCK_PROFILE


# ── Existing section lookup ────────────────────────────────────────────────────

async def _get_existing_section_text(
    db: AsyncSession, report_id: str, section_code: str
) -> str | None:
    """Fetch the current text of a report section from the database.

    Args:
        db: Active async session.
        report_id: Parent report UUID.
        section_code: RICS section code (e.g. ``"E4"`` for Main walls).

    Returns:
        Section text string, or ``None`` if no section exists yet.
    """
    result = await db.execute(
        select(ReportSection).where(
            ReportSection.report_id == report_id,
            ReportSection.section_code == section_code,
        )
    )
    section_orm = result.scalars().first()
    return section_orm.text if section_orm else None


# ── Main entry point ───────────────────────────────────────────────────────────


def validate_section_codes(section_codes: list[str]) -> None:
    """Raise ``ValueError`` when any code is not a known RICS section."""
    invalid = [c for c in section_codes if c not in ALL_VALID_SECTION_CODES]
    if invalid:
        raise ValueError(
            f"Unknown section code(s): {', '.join(invalid)}. "
            "Use valid RICS template section codes."
        )


async def mark_report_generation_failed(
    report_id: str,
    tenant_id: str,
    error_message: str,
) -> None:
    """Atomically move a report out of ``generating`` into ``failed``."""
    msg = (error_message or "Generation failed.")[:2000]
    factory = get_session_factory()
    async with factory() as db:
        result = await db.execute(
            update(Report)
            .where(
                Report.id == report_id,
                Report.tenant_id == tenant_id,
                Report.status == ReportStatus.generating,
            )
            .values(
                status=ReportStatus.failed,
                generation_started_at=None,
                generation_section_total=None,
                error_message=msg,
            )
        )
        if result.rowcount == 0:
            report = await db.get(Report, report_id)
            if (
                report is not None
                and report.tenant_id == tenant_id
                and report.status == ReportStatus.generating
            ):
                report.status = ReportStatus.failed
                report.generation_started_at = None
                report.generation_section_total = None
                report.error_message = msg
        await db.commit()


async def mark_report_generation_complete_if_still_generating(
    report_id: str,
    tenant_id: str,
) -> None:
    """Safety net when section work finished but report status was not finalized."""
    factory = get_session_factory()
    async with factory() as db:
        await db.execute(
            update(Report)
            .where(
                Report.id == report_id,
                Report.tenant_id == tenant_id,
                Report.status == ReportStatus.generating,
            )
            .values(
                status=ReportStatus.complete,
                generation_started_at=None,
                generation_section_total=None,
                error_message=None,
            )
        )
        await db.commit()


async def finalize_generation_status_if_stuck(
    report_id: str,
    tenant_id: str,
    *,
    error_message: str = "Generation ended without updating report status.",
) -> None:
    """If a job exits while the report is still ``generating``, mark it ``failed``."""
    await mark_report_generation_failed(report_id, tenant_id, error_message)


async def abort_generation_if_report_invalid(
    report_id: str,
    tenant_id: str,
    *,
    reason: str = "Report not found or access denied.",
) -> bool:
    """Return False and mark the report failed when it cannot be generated."""
    factory = get_session_factory()
    async with factory() as db:
        report = await db.get(Report, report_id)
        if report is None:
            return False
        if report.tenant_id != tenant_id:
            await mark_report_generation_failed(
                report_id,
                report.tenant_id,
                reason,
            )
            return False
    return True


async def report_generation_still_active(report_id: str, tenant_id: str) -> bool:
    """True while the report is still in ``generating`` (not cancelled/finalized)."""
    factory = get_session_factory()
    async with factory() as db:
        report = await db.get(Report, report_id)
        return (
            report is not None
            and report.tenant_id == tenant_id
            and report.status == ReportStatus.generating
        )


async def run_generation(
    report_id: str,
    tenant_id: str,
    template_id: str,
    bullets: list[str],
    mode: str = GenerationMode.generate,
    ai_level: int = AILevel.balanced,
    ai_percent: int | None = None,
    retrieval_level: str = "paragraph",
    force_regenerate: bool = False,
    strict_uploaded_only: bool = False,
    reference_document_ids: list[str] | None = None,
    draft_paragraph: str | None = None,
    interference_level: str | None = None,
    template_ids: list[str] | None = None,
    bullets_by_section: dict[str, list[str]] | None = None,
) -> None:
    """Run the full AI pipeline for one or more report sections.

    When ``template_ids`` lists multiple section codes, all are processed for
    ``generate``, ``proofread``, and ``enhance``. With ``enable_async_pipeline``,
    work runs in parallel; otherwise sections run sequentially.

    Dispatches to one of three sub-pipelines based on ``mode``:

    * ``generate`` — RAG + style-personalised generation (stages 1–9).
    * ``proofread`` — Retrieve existing text, proofread, re-persist.
    * ``enhance``   — Retrieve existing text + more evidence, expand, re-persist.
    """
    try:
        await _run_generation_body(
            report_id=report_id,
            tenant_id=tenant_id,
            template_id=template_id,
            bullets=bullets,
            mode=mode,
            ai_level=ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            force_regenerate=force_regenerate,
            strict_uploaded_only=strict_uploaded_only,
            reference_document_ids=reference_document_ids,
            draft_paragraph=draft_paragraph,
            interference_level=interference_level,
            template_ids=template_ids,
            bullets_by_section=bullets_by_section,
        )
    except Exception as exc:
        logger.exception("Unhandled error in run_generation report=%s", report_id)
        await mark_report_generation_failed(report_id, tenant_id, str(exc))
    finally:
        await finalize_generation_status_if_stuck(report_id, tenant_id)


async def _run_generation_body(
    report_id: str,
    tenant_id: str,
    template_id: str,
    bullets: list[str],
    mode: str,
    ai_level: int,
    ai_percent: int | None,
    retrieval_level: str,
    force_regenerate: bool,
    strict_uploaded_only: bool,
    reference_document_ids: list[str] | None,
    draft_paragraph: str | None,
    interference_level: str | None,
    template_ids: list[str] | None,
    bullets_by_section: dict[str, list[str]] | None,
) -> None:
    sections = _resolve_generation_sections(template_id, template_ids)
    try:
        validate_section_codes(sections)
    except ValueError as exc:
        logger.warning(
            "Invalid section codes for report=%s: %s",
            report_id,
            exc,
        )
        await mark_report_generation_failed(report_id, tenant_id, str(exc))
        return
    if not await abort_generation_if_report_invalid(report_id, tenant_id):
        return
    if len(sections) > 1:
        multi_kwargs = dict(
            mode=mode,
            report_id=report_id,
            tenant_id=tenant_id,
            section_codes=sections,
            bullets=bullets,
            bullets_by_section=bullets_by_section,
            ai_level=ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            force_regenerate=force_regenerate,
            strict_uploaded_only=strict_uploaded_only,
            reference_document_ids=reference_document_ids,
            draft_paragraph=draft_paragraph,
            interference_level=interference_level,
        )
        try:
            from app.db.database import effective_section_concurrency

            concurrency = effective_section_concurrency()
            if concurrency > 1:
                await _run_multi_section_parallel(
                    **multi_kwargs, concurrency=concurrency
                )
            else:
                await _run_multi_section_sequential(**multi_kwargs)
        except Exception as exc:
            logger.exception(
                "Multi-section generation failed report=%s mode=%s",
                report_id,
                mode,
            )
            await mark_report_generation_failed(report_id, tenant_id, str(exc))
        return

    refs = list(reference_document_ids or [])
    tier_norm = _normalise_interference_level(interference_level)
    sec = sections[0]
    sec_bullets = _bullets_for_section(sec, bullets, bullets_by_section)
    factory = get_session_factory()
    async with factory() as db:
        report: Report | None = await db.get(Report, report_id)
        if report is None or report.tenant_id != tenant_id:
            await mark_report_generation_failed(
                report_id,
                tenant_id,
                "Report not found or access denied.",
            )
            return

        try:
            if mode == GenerationMode.proofread:
                await _run_proofread(
                    db,
                    report,
                    tenant_id,
                    sec,
                    sec_bullets,
                    ai_level,
                    ai_percent=ai_percent,
                    retrieval_level=retrieval_level,
                    reference_document_ids=refs,
                    interference_level=tier_norm,
                )
            elif mode == GenerationMode.enhance:
                await _run_enhance(
                    db,
                    report,
                    tenant_id,
                    sec,
                    sec_bullets,
                    force_regenerate,
                    ai_level,
                    ai_percent=ai_percent,
                    retrieval_level=retrieval_level,
                    reference_document_ids=refs,
                    interference_level=tier_norm,
                )
            else:
                await _run_generate_primary(
                    db=db,
                    report=report,
                    tenant_id=tenant_id,
                    template_id=sec,
                    bullets=sec_bullets,
                    force_regenerate=force_regenerate,
                    ai_level=ai_level,
                    ai_percent=ai_percent,
                    retrieval_level=retrieval_level,
                    strict_uploaded_only=strict_uploaded_only,
                    reference_document_ids=refs,
                    draft_paragraph=draft_paragraph,
                    interference_level=tier_norm,
                )

        except Exception as exc:
            logger.exception("Generation failed for report=%s mode=%s", report_id, mode)
            report.status = ReportStatus.failed
            report.generation_started_at = None
            report.generation_section_total = None
            report.error_message = str(exc)[:2000]
            await db.commit()
        else:
            await mark_report_generation_complete_if_still_generating(
                report_id,
                tenant_id,
            )


def _resolve_generation_sections(
    template_id: str,
    template_ids: list[str] | None,
) -> list[str]:
    if template_ids:
        seen: set[str] = set()
        out: list[str] = []
        for code in [template_id, *template_ids]:
            c = str(code).strip()
            if c and c not in seen:
                seen.add(c)
                out.append(c)
        return out if out else [template_id]
    return [template_id]


def _bullets_for_section(
    section_code: str,
    bullets: list[str],
    bullets_by_section: dict[str, list[str]] | None,
) -> list[str]:
    if bullets_by_section and section_code in bullets_by_section:
        return list(bullets_by_section[section_code])
    return list(bullets)


def _finalize_multi_section_report(
    report: Report,
    *,
    failure_count: int,
    total: int,
    phase: str,
) -> None:
    """Set report status after a multi-section job (generate / proofread / enhance)."""
    report.generation_started_at = None
    report.generation_section_total = None
    successes = total - failure_count
    if successes == 0:
        report.status = ReportStatus.failed
        if not report.error_message:
            report.error_message = f"All {total} sections failed during {phase}."
    elif failure_count:
        report.status = ReportStatus.partial
        report.error_message = (
            f"{failure_count} of {total} sections failed during {phase}; "
            f"{successes} succeeded."
        )
    else:
        report.status = ReportStatus.complete
        report.error_message = None


async def _run_section_job(
    *,
    mode: str,
    db: AsyncSession,
    report: Report,
    tenant_id: str,
    section_code: str,
    sec_bullets: list[str],
    ai_level: int,
    ai_percent: int | None,
    retrieval_level: str,
    force_regenerate: bool,
    strict_uploaded_only: bool,
    reference_document_ids: list[str] | None,
    draft_paragraph: str | None,
    interference_level: str | None,
) -> None:
    """Run one section pipeline without marking the report complete."""
    tier_norm = _normalise_interference_level(interference_level)
    refs = list(reference_document_ids or [])
    if mode == GenerationMode.proofread:
        await _run_proofread(
            db,
            report,
            tenant_id,
            section_code,
            sec_bullets,
            ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            reference_document_ids=refs,
            interference_level=tier_norm,
            mark_report_complete=False,
        )
    elif mode == GenerationMode.enhance:
        await _run_enhance(
            db,
            report,
            tenant_id,
            section_code,
            sec_bullets,
            force_regenerate,
            ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            reference_document_ids=refs,
            interference_level=tier_norm,
            mark_report_complete=False,
        )
    else:
        await _run_generate_primary(
            db=db,
            report=report,
            tenant_id=tenant_id,
            template_id=section_code,
            bullets=sec_bullets,
            force_regenerate=force_regenerate,
            ai_level=ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            strict_uploaded_only=strict_uploaded_only,
            reference_document_ids=refs,
            draft_paragraph=draft_paragraph,
            interference_level=tier_norm,
            mark_report_complete=False,
        )


async def _run_multi_section_parallel(
    *,
    mode: str,
    report_id: str,
    tenant_id: str,
    section_codes: list[str],
    bullets: list[str],
    bullets_by_section: dict[str, list[str]] | None,
    ai_level: int,
    ai_percent: int | None,
    retrieval_level: str,
    force_regenerate: bool,
    strict_uploaded_only: bool,
    reference_document_ids: list[str] | None,
    draft_paragraph: str | None,
    interference_level: str | None,
    concurrency: int = 0,
) -> None:
    """Run multiple sections concurrently, bounded by ``concurrency``.

    The global LLM throttle (``max_concurrent_llm_calls``) still caps in-flight
    provider calls; this semaphore bounds how many sections hold a DB session and
    retrieval state at once, which keeps SQLite write contention and memory in check.
    """
    import structlog

    log = structlog.get_logger(__name__)
    if not await abort_generation_if_report_invalid(report_id, tenant_id):
        return

    limit = concurrency if concurrency and concurrency > 0 else len(section_codes)
    sem = asyncio.Semaphore(max(1, limit))

    async def _one(section_code: str) -> None:
        async with sem:
            sec_bullets = _bullets_for_section(section_code, bullets, bullets_by_section)
            factory = get_session_factory()
            async with factory() as db:
                report = await db.get(Report, report_id)
                if report is None:
                    raise RuntimeError(
                        f"Report {report_id} not found during parallel section {section_code}",
                    )
                await _run_section_job(
                    mode=mode,
                    db=db,
                    report=report,
                    tenant_id=tenant_id,
                    section_code=section_code,
                    sec_bullets=sec_bullets,
                    ai_level=ai_level,
                    ai_percent=ai_percent,
                    retrieval_level=retrieval_level,
                    force_regenerate=force_regenerate,
                    strict_uploaded_only=strict_uploaded_only,
                    reference_document_ids=reference_document_ids,
                    draft_paragraph=draft_paragraph,
                    interference_level=interference_level,
                )

    results = await asyncio.gather(
        *[_one(code) for code in section_codes],
        return_exceptions=True,
    )
    failed_codes: list[str] = []
    for code, result in zip(section_codes, results, strict=True):
        if isinstance(result, BaseException):
            failed_codes.append(code)
            log.error(
                event="multi_section_parallel_section_failed",
                phase=mode,
                section_id=code,
                error=str(result),
                exc_type=type(result).__name__,
            )
    failure_count = len(failed_codes)
    if failure_count:
        log.warning(
            event="multi_section_parallel_partial_failure",
            phase=mode,
            failed=failure_count,
            failed_sections=failed_codes,
            total=len(section_codes),
        )
    factory = get_session_factory()
    async with factory() as db:
        report = await db.get(Report, report_id)
        if report is None:
            await mark_report_generation_failed(
                report_id,
                tenant_id,
                "Report not found during multi-section finalization.",
            )
            return
        _finalize_multi_section_report(
            report,
            failure_count=failure_count,
            total=len(section_codes),
            phase=mode,
        )
        await db.commit()


async def _run_multi_section_sequential(
    *,
    mode: str,
    report_id: str,
    tenant_id: str,
    section_codes: list[str],
    bullets: list[str],
    bullets_by_section: dict[str, list[str]] | None,
    ai_level: int,
    ai_percent: int | None,
    retrieval_level: str,
    force_regenerate: bool,
    strict_uploaded_only: bool,
    reference_document_ids: list[str] | None,
    draft_paragraph: str | None,
    interference_level: str | None,
) -> None:
    """Run multiple sections sequentially (async pipeline off)."""
    factory = get_session_factory()
    if not await abort_generation_if_report_invalid(report_id, tenant_id):
        return
    failures = 0
    processed = 0
    for section_code in section_codes:
        if not await report_generation_still_active(report_id, tenant_id):
            logger.info(
                "Sequential generation stopped report=%s before section=%s (no longer generating)",
                report_id,
                section_code,
            )
            failures += len(section_codes) - processed
            break
        sec_bullets = _bullets_for_section(section_code, bullets, bullets_by_section)
        async with factory() as db:
            report = await db.get(Report, report_id)
            if report is None:
                await mark_report_generation_failed(
                    report_id,
                    tenant_id,
                    "Report not found during sequential multi-section run.",
                )
                return
            try:
                await _run_section_job(
                    mode=mode,
                    db=db,
                    report=report,
                    tenant_id=tenant_id,
                    section_code=section_code,
                    sec_bullets=sec_bullets,
                    ai_level=ai_level,
                    ai_percent=ai_percent,
                    retrieval_level=retrieval_level,
                    force_regenerate=force_regenerate,
                    strict_uploaded_only=strict_uploaded_only,
                    reference_document_ids=reference_document_ids,
                    draft_paragraph=draft_paragraph,
                    interference_level=interference_level,
                )
            except Exception:
                failures += 1
                logger.exception(
                    "Sequential multi-section failed report=%s section=%s mode=%s",
                    report_id,
                    section_code,
                    mode,
                )
        processed += 1

    async with factory() as db:
        report = await db.get(Report, report_id)
        if report is None:
            await mark_report_generation_failed(
                report_id,
                tenant_id,
                "Report not found during sequential multi-section finalization.",
            )
            return
        _finalize_multi_section_report(
            report,
            failure_count=failures,
            total=len(section_codes),
            phase=mode,
        )
        await db.commit()


# ── Pure text-generation helper (no DB writes, no status change) ───────────────

def _normalise_ai_controls(ai_level: int, ai_percent: int | None) -> tuple[int, int, bool]:
    """Normalise AI controls to (legacy_level_1_to_5, percent_0_to_100, rag_only_bool)."""
    lvl = max(1, min(5, int(ai_level)))
    p = ai_level_to_percent(lvl)
    if ai_percent is not None:
        try:
            p = int(ai_percent)
        except Exception:  # noqa: BLE001
            p = p
        p = max(0, min(100, p))
        lvl = ai_percent_to_level(p)
    rag_only = p <= 5
    return lvl, p, rag_only


def _ai_level_to_params(ai_level: int, ai_percent: int | None = None) -> dict[str, Any]:
    """Translate AI intensity into adapter kwargs and prompt hints.

    Preferred control is ``ai_percent`` (0–100). ``ai_level`` is supported for
    backward compatibility and is mapped to 0–100 in 25-point increments.
    """
    level, pct, rag_only = _normalise_ai_controls(ai_level, ai_percent)
    # Temperature: ~0 at 0% (assembly), ~0.42 at 100% — low end is nearly deterministic.
    temperature = round(0.0 + 0.42 * (pct / 100.0) ** 1.15, 3)
    if pct <= 12:
        temperature = 0.0

    # How freely the LLM may bridge / paraphrase beyond the raw bullets
    if pct <= 12:
        creativity_hint = (
            "ASSEMBLY MODE (0–12%): You are a TEMPLATE ASSEMBLER. Every clause in your output MUST be "
            "a verbatim quote from a STANDARD SOURCE PASSAGE, the SECTION SKELETON, or the RAW NOTES. "
            "DO NOT paraphrase. DO NOT replace any source word with a synonym (technical or otherwise). "
            "DO NOT reorder clauses 'for flow' or tighten 'for clarity'. "
            "Allowed new words: at most 12 short connectors across the whole output (and/but/however/"
            "Additionally/The/This), UK-spelling corrections of American spellings, and property-specific "
            "values lifted from RAW NOTES. If retrieved passages do not cover a subsection, skip it — "
            "do not fill with original prose."
        )
    elif pct <= 37:
        creativity_hint = (
            "LOW INVOLVEMENT (13–37%): Quote STANDARD SOURCE PASSAGES verbatim by default. "
            "Edits permitted only for grammar/tense or to drop an inapplicable clause. "
            "DO NOT replace technical terms or standard phrases with synonyms. "
            "New prose limited to short bridging sentences (under 15 words) linking two source passages."
        )
    elif pct <= 67:
        creativity_hint = (
            "MODERATE INVOLVEMENT (38–67%): Adapt tone and flow while preserving facts and "
            "the intent of standard paragraphs. Limited original bridging."
        )
    elif pct <= 87:
        creativity_hint = (
            "HIGH INVOLVEMENT (68–87%): Strong style adaptation and fluent prose; facts still "
            "grounded in notes and retrieved evidence."
        )
    else:
        creativity_hint = (
            "MAXIMUM INVOLVEMENT (88–100%): Full professional drafting — summarise, expand, "
            "restructure as needed; still no invented property-specific facts."
        )

    # Whether to skip notes expansion at rag-only intensity (pure RAG pass-through)
    skip_expansion = rag_only or level == 1 or pct <= 12
    assembly_mode = pct <= 12
    return {
        "temperature": temperature,
        "creativity_hint": creativity_hint,
        "skip_expansion": skip_expansion,
        "ai_percent": pct,
        "ai_level": level,
        "assembly_mode": assembly_mode,
    }


def _mode_controls(ai_level: int, mode: str, ai_percent: int | None = None) -> tuple[float, str]:
    """Return mode-specific ``(temperature, creativity_hint)`` for ai_level.

    ``generate`` uses the base mapping directly.
    ``proofread`` is intentionally a little more conservative than generate.
    ``enhance`` allows a slightly richer style than generate.
    """
    base = _ai_level_to_params(ai_level, ai_percent=ai_percent)
    t = float(base["temperature"])
    h = str(base["creativity_hint"])

    if mode == GenerationMode.proofread:
        # Keep edits controlled even at higher AI levels.
        temperature = max(0.05, round(t - 0.04, 3))
        hint = (
            "PROOFREAD MODE INTENSITY: "
            + h
            + " Focus on language quality and style alignment only; do not alter factual meaning."
        )
        return temperature, hint

    if mode == GenerationMode.enhance:
        # Permit slightly more fluency/bridging for technical expansion.
        temperature = min(0.5, round(t + 0.03, 3))
        hint = "ENHANCE MODE INTENSITY: " + h
        return temperature, hint

    return t, h


async def _tenant_report_source_doc_ids(
    db: AsyncSession | None,
    tenant_id: str,
) -> list[str]:
    """Return all of the tenant's successfully-ingested report-source doc IDs.

    Used to widen RAG retrieval to the tenant's whole library (old + new
    uploads) when ``settings.rag_use_full_tenant_library`` is enabled.
    ``style_corpus`` documents are excluded so past-report observations never
    enter the factual evidence pool. Best-effort: returns ``[]`` on any error so
    generation degrades to per-report isolation rather than failing.
    """
    if db is None:
        return []
    try:
        from sqlalchemy import select as _select

        from app.db.models import Document, DocumentPurpose, IngestStatus

        result = await db.execute(
            _select(Document.id).where(
                Document.tenant_id == tenant_id,
                Document.status == IngestStatus.complete,
                Document.document_purpose == DocumentPurpose.report_source,
            )
        )
        return [str(row[0]) for row in result.all() if row[0]]
    except Exception as exc:  # noqa: BLE001 — never break generation on this widening step
        logger.warning("Full-library doc id lookup failed tenant=%s: %s", tenant_id, exc)
        return []


async def _generate_section_text(
    tenant_id: str,
    template_id: str,
    bullets: list[str],
    style_profile: WritingStyleProfile,
    ai_level: int = AILevel.balanced,
    ai_percent: int | None = None,
    primary_document_id: str | None = None,
    reference_document_ids: list[str] | None = None,
    draft_paragraph: str | None = None,
    report_id: str | None = None,
    retrieval_level: str = "paragraph",
    *,
    db: AsyncSession | None = None,
    report_survey_level: int | None = None,
    strict_uploaded_only: bool = False,
    interference_level: str | None = None,
) -> tuple[str, list[dict[str, Any]], float, list[RerankedResult], list[SearchResult], dict[str, Any]]:
    """Run the RAG generation pipeline and return ``(text, provenance, confidence, top_results)``.

    This helper intentionally performs **no database writes** and does **not**
    touch ``report.status``.  It is used both by :func:`_run_generate` (which
    persists afterwards) and as a seed-text fallback inside
    :func:`_run_proofread` and :func:`_run_enhance` — where persisting
    prematurely would mark the report ``complete`` and cause the frontend to
    load generate-mode output before the actual proofread/enhance call runs.

    Args:
        tenant_id: Owning tenant for retrieval isolation.
        template_id: RICS section code.
        bullets: User-supplied fact bullets.
        style_profile: Pre-built writing style profile.
        ai_level: 1–5 AI interference level.
        primary_document_id: Prefer chunks from this uploaded document (report source file).
        reference_document_ids: Additional uploads to prioritise (exemplar reports).
        draft_paragraph: Optional paragraph to mirror for tone/structure.
        report_id: When set, includes runtime-edited section index (``runtime-{id}-{section}``) in hierarchical retrieval.
        retrieval_level: Strict retrieval granularity: document | section | paragraph.

    Returns:
        5-tuple ending with document+section context rows used for provenance (excludes paragraph rerank duplicates handled in-loop).
    """
    level_params = _ai_level_to_params(ai_level, ai_percent=ai_percent)
    # Hard guarantee: when notes_only_generation is enabled, do NOT allow RAG
    # uploads to contribute property-specific facts. Retrieval is still allowed
    # as reference-only guidance (terminology/structure), with provenance suppressed.
    notes_only = bool(getattr(settings, "notes_only_generation", True))
    ref_ids: list[str] = list(reference_document_ids or [])

    async def _apply_survey_filter(pool: list[SearchResult]) -> list[SearchResult]:
        if db is None or report_survey_level is None:
            return pool
        return await filter_search_results_by_survey_level(
            db,
            tenant_id=tenant_id,
            results=pool,
            report_survey_level=report_survey_level,
            primary_document_id=primary_document_id,
            reference_document_ids=ref_ids,
        )

    bullet_cap = int(settings.max_bullets_per_section)
    raw_note_lines = list(bullets or [])
    from app.generator.note_bullets import clean_and_clamp_bullets_with_report

    bullets, clamp_report = clean_and_clamp_bullets_with_report(
        raw_note_lines, max_items=bullet_cap
    )
    if clamp_report.has_clamps:
        logger.warning(
            "Section %s notes clamped: input=%d kept=%d duplicate_dropped=%d "
            "overflow_dropped=%d (cap=%d). Surfaced in metrics['bullet_clamps'].",
            template_id,
            clamp_report.input_count,
            clamp_report.cleaned_count,
            clamp_report.duplicate_dropped,
            clamp_report.overflow_dropped,
            bullet_cap,
        )
    bullets_for_generation = bullets

    # Fetch photo observations up-front so a section that has photos but no
    # typed notes can still be generated FROM the photo evidence (instead of
    # returning blank). Vision runs at most once; the value is reused below.
    _early_photo_obs: list[str] = []
    _early_photo_note: str | None = None
    try:
        _early_photo_obs, _early_photo_note = await get_photo_observations_for_section(
            db,
            tenant_id=tenant_id,
            report_id=report_id,
            section_code=template_id,
        )
    except Exception:  # noqa: BLE001 — never block generation on photo analysis
        _early_photo_obs, _early_photo_note = [], None

    if not bullets_for_generation and not _early_photo_obs:
        # No notes AND no usable photo evidence: persist a blank section so the
        # UI can prompt the user to fill it in immediately (inline editor).
        empty_metrics: dict[str, Any] = {
            "requested_ai_percent": int(level_params.get("ai_percent", 50))
        }
        if clamp_report.has_clamps:
            empty_metrics["bullet_clamps"] = clamp_report.to_dict()
        if _early_photo_note:
            empty_metrics["photo_note"] = _early_photo_note
        return "", [], 0.0, [], [], empty_metrics

    async def _expand_bullets_coro() -> list[str]:
        """Notes expansion; OpenAI path runs in a thread so the event loop stays responsive."""
        if not bullets_for_generation:
            # Photo-only section: nothing to expand; photo evidence is appended below.
            return []
        if level_params["skip_expansion"]:
            logger.debug("AI level 1 (RAG only): skipping notes expansion for section=%s", template_id)
            return list(bullets_for_generation)
        key = (settings.openai_api_key or "").strip()
        if not key:
            out = expand_notes(
                bullets=bullets,
                section_code=template_id,
                openai_api_key="",
                chat_model=settings.chat_model,
            )
            logger.debug(
                "Notes expansion (rule-based): %d bullets → %d for section=%s",
                len(bullets_for_generation),
                len(out),
                template_id,
            )
            return out
        out = await expand_notes_async(
            bullets=bullets,
            section_code=template_id,
            openai_api_key=key,
            chat_model=settings.chat_model,
        )
        logger.debug(
            "Notes expansion: %d bullets → %d expanded for section=%s",
            len(bullets_for_generation),
            len(out),
            template_id,
        )
        return out

    async def _photo_policy_coro() -> tuple[PhotoPolicy, str]:
        """Photo policy classification (vision observations already fetched above)."""
        return await classify_section_photo_policy_async(
            db, tenant_id, template_id, survey_level=report_survey_level
        )

    expanded_bullets, policy_row = await asyncio.gather(
        _expand_bullets_coro(),
        _photo_policy_coro(),
    )
    (photo_policy, _) = policy_row
    # Reuse the observations fetched before the empty-notes gate (vision runs once).
    photo_obs, photo_note = _early_photo_obs, _early_photo_note

    if photo_policy == PhotoPolicy.requires_image and not photo_obs and not photo_note:
        expanded_bullets = list(expanded_bullets) + [
            "We were unable to verify this during inspection. "
            "No photos were provided for this section; confirm condition via site inspection or photo evidence before sign-off."
        ]
    elif photo_note:
        expanded_bullets = list(expanded_bullets) + [f"Photo policy note: {photo_note}"]
    elif photo_obs:
        expanded_bullets = (
            list(expanded_bullets)
            + ["Photographic evidence (observed in the inspection photos for this section):"]
            + [f"- {x}" for x in photo_obs]
        )

    # Photo-derived observations are first-class inspection EVIDENCE — they must
    # be treated as a trusted source by the anti-hallucination grounding pass,
    # otherwise the verifier strips every photo-derived sentence as "ungrounded"
    # and the photos never visibly contribute to the section. ``bullets`` stays
    # the strict grounding key for notes; ``grounding_evidence`` augments it with
    # the vision observations only.
    grounding_evidence: list[str] = list(bullets) + list(photo_obs or [])

    # Use expanded bullets for internal query strings (and retrieval when enabled).
    query = " ".join(expanded_bullets)[:12_000]
    pack = get_survey_pack(report_survey_level)
    template = get_template(template_id, report_survey_level)
    skeleton = template.skeleton if template else f"[{template_id}]: [content]."
    if int(report_survey_level or 3) >= 1 and template_id not in ("A", "B", "C", "L") and template is not None:
        skeleton = _wrap_structured_skeleton(
            survey_level=int(report_survey_level or 3),
            code=template_id,
            title=template.title,
            base_skeleton=skeleton,
            has_condition_rating=bool(getattr(template, "has_condition_rating", False)),
        )

    runtime_extra: list[str] = []
    if report_id and template_id:
        runtime_extra = [runtime_section_vector_doc_id(str(report_id), template_id)]

    rl = str(retrieval_level or "paragraph").strip().lower()
    if rl not in ("document", "section", "paragraph"):
        rl = "paragraph"

    ap_inv = int(level_params.get("ai_percent", 50))
    _rerank_boost = 5 if ap_inv <= 12 else (2 if ap_inv <= 37 else 0)
    rerank_n = min(10, int(getattr(settings, "rerank_top_n", 3)) + _rerank_boost)

    # Retrieval. In notes-only mode, retrieved snippets are treated as *reference-only*
    # guidance (terminology/structure) and redacted; they are never surfaced as
    # provenance and never treated as factual evidence.
    doc_snippets: list[str] = []
    hierarchy_sec_snippets: list[str] = []
    para_snippets: list[str] = []
    doc_ctx_results: list[SearchResult] = []
    top_results: list[RerankedResult] = []

    # Per-report isolation: retrieval draws from documents explicitly attached
    # to this report. In strict_uploaded_only mode we *also* disallow KB/Behrang
    # corpora and any boilerplate sources outside the attached doc set.
    universe: set[str] = set()
    if primary_document_id:
        universe.add(str(primary_document_id))
    universe.update(ref_ids)
    universe.update(runtime_extra)

    # Whole-library retrieval (old + new uploads). When enabled and we are NOT
    # in strict per-report isolation, widen the admissible document set to every
    # ingested report-source document for this tenant. The report's own upload
    # (primary_document_id) and explicit references stay prioritised by the
    # retriever's ordering; style_corpus is excluded by the query above and by
    # the report-source filter in the lookup helper. Runtime section vectors are
    # always kept.
    library_doc_ids: list[str] = []
    if (
        getattr(settings, "rag_use_full_tenant_library", True)
        and not strict_uploaded_only
    ):
        library_doc_ids = await _tenant_report_source_doc_ids(db, tenant_id)
        if library_doc_ids:
            universe.update(library_doc_ids)
            logger.info(
                "Full-library retrieval for section=%s tenant=%s: %d report-source "
                "doc(s) admissible (primary=%s, explicit_refs=%d)",
                template_id,
                tenant_id,
                len(library_doc_ids),
                (primary_document_id[:8] if primary_document_id else None),
                len(ref_ids),
            )

    allowed = frozenset(universe) if universe else None

    if not allowed:
        logger.info(
            "Strict isolation: no documents attached to report — skipping retrieval for section=%s. "
            "Generation will use bullets only.",
            template_id,
        )
    elif settings.hierarchical_rag_enabled:
        vs = get_vectorstore()
        if rl == "document":
            sk = (skeleton or "").strip().replace("\n", " ")[:500]
            broad = (
                f"{pack.product_label} section {template_id}. "
                f"Whole document scope and narrative. {sk} {query[:400]}"
            )
            if use_async_retrieval_path():
                hits = await retrieve_scoped_async(
                    broad,
                    tenant_id,
                    k=max(settings.hierarchical_k_document * 25, 40),
                    hierarchy_level="document",
                    doc_id_in=allowed,
                )
            else:
                hits = vs.search(
                    broad,
                    tenant_id,
                    k=max(settings.hierarchical_k_document * 25, 40),
                    hierarchy_level="document",
                    doc_id_in=allowed,
                )
            hits = await _apply_survey_filter(hits)
            hits = [h for h in hits if str(getattr(h, "doc_id", "")) in allowed]
            doc_ctx_results = hits[: settings.hierarchical_k_document]
            doc_snippets = [r.text for r in doc_ctx_results]
        elif rl == "section":
            if use_async_retrieval_path():
                hits = await retrieve_scoped_async(
                    query,
                    tenant_id,
                    k=max(settings.hierarchical_k_section * 30, 60),
                    hierarchy_level="section",
                    doc_id_in=allowed,
                )
            else:
                hits = vs.search(
                    query,
                    tenant_id,
                    k=max(settings.hierarchical_k_section * 30, 60),
                    hierarchy_level="section",
                    doc_id_in=allowed,
                )
            hits = await _apply_survey_filter(hits)
            hits = [h for h in hits if str(getattr(h, "doc_id", "")) in allowed]
            doc_ctx_results = hits[: settings.hierarchical_k_section]
            hierarchy_sec_snippets = [r.text for r in doc_ctx_results]
        else:  # paragraph
            if use_async_retrieval_path():
                hits = await retrieve_scoped_async(
                    query,
                    tenant_id,
                    k=max(settings.hierarchical_k_paragraph_pool * 25, 80),
                    hierarchy_level="paragraph",
                    doc_id_in=allowed,
                )
            else:
                hits = vs.search(
                    query,
                    tenant_id,
                    k=max(settings.hierarchical_k_paragraph_pool * 25, 80),
                    hierarchy_level="paragraph",
                    doc_id_in=allowed,
                )
            hits = await _apply_survey_filter(hits)
            hits = [h for h in hits if str(getattr(h, "doc_id", "")) in allowed]
            top_results = rerank(query=query, results=hits, top_n=rerank_n)
            # At assembly/low tier, prefer chunks tagged as boilerplate so the
            # firm's approved standard wording lands at the top of context.
            if ap_inv <= 37:
                top_results = reorder_by_chunk_role(top_results)
            para_snippets = [r.text for r in top_results]
    else:
        # Legacy index path. The per-doc-id filter inside vs.search() is not
        # honoured here, so we filter results post-hoc against the report's
        # attached document set.
        if rl == "document":
            if use_async_retrieval_path():
                doc_ctx_results = await retrieve_document_level_context_async(
                    template_id=template_id,
                    skeleton_excerpt=skeleton,
                    tenant_id=tenant_id,
                    primary_document_id=primary_document_id,
                    reference_document_ids=ref_ids,
                    product_label=pack.product_label,
                )
            else:
                from functools import partial

                from app.async_executor import run_sync_in_executor

                doc_ctx_results = await run_sync_in_executor(
                    partial(
                        retrieve_document_level_context,
                        template_id=template_id,
                        skeleton_excerpt=skeleton,
                        tenant_id=tenant_id,
                        primary_document_id=primary_document_id,
                        reference_document_ids=ref_ids,
                        product_label=pack.product_label,
                    )
                )
            doc_ctx_results = await _apply_survey_filter(doc_ctx_results)
            doc_ctx_results = [r for r in doc_ctx_results if str(getattr(r, "doc_id", "")) in allowed]
            doc_snippets = [r.text for r in doc_ctx_results]
        elif rl == "section":
            doc_ctx_results = []
            hierarchy_sec_snippets = []
        else:
            candidates = await retrieve_for_report_unified(
                query=query,
                tenant_id=tenant_id,
                primary_document_id=primary_document_id,
                secondary_document_ids=ref_ids,
            )
            candidates = await _apply_survey_filter(candidates)
            candidates = [c for c in candidates if str(getattr(c, "doc_id", "")) in allowed]
            top_results = rerank(query=query, results=candidates, top_n=rerank_n)
            if ap_inv <= 37:
                top_results = reorder_by_chunk_role(top_results)
            para_snippets = [r.text for r in top_results]

    if notes_only:
        # Assembly mode (very low AI %): keep retrieved wording intact so the model can
        # reuse standard / boilerplate phrasing; higher modes redact fact-shaped tokens.
        if not level_params.get("assembly_mode"):
            doc_snippets = _redact_reference_snippets(doc_snippets)
            hierarchy_sec_snippets = _redact_reference_snippets(hierarchy_sec_snippets)
            para_snippets = _redact_reference_snippets(para_snippets)

    # Firm standard paragraphs: master .docx under knowledge_base_dirs (e.g. Behrang RICS Documents).
    #
    # Special rule for true 0%: only use the section's STANDARD PARAGRAPH (plus bullets),
    # never any other retrieved content. This keeps the 0% tier behaving like strict
    # template assembly, even when the report has attached PDFs that would otherwise
    # contribute additional wording.
    section_std_raw = (get_standard_paragraph_for_section(template_id) or "").strip()
    # Hybrid mode: the firm's own standard paragraph is a TEMPLATE (placeholders
    # + option brackets, no real property facts), so it is exempt from notes-only
    # redaction and is used to shape wording at every involvement tier. RAG
    # uploads from other reports stay redacted (anti-leak preserved).
    shape_with_std = bool(getattr(settings, "standard_paragraphs_shape_wording", True))
    section_std_prepared = ""
    if section_std_raw:
        if notes_only and not shape_with_std:
            section_std_prepared = _redact_reference_snippet(section_std_raw).strip()
        else:
            section_std_prepared = section_std_raw
    if int(level_params.get("ai_percent", 50)) == 0:
        if section_std_prepared:
            para_snippets = [section_std_prepared]
            doc_snippets = []
            hierarchy_sec_snippets = []
        else:
            # No standard paragraph for this section — do NOT substitute other retrieved text.
            para_snippets = []
            doc_snippets = []
            hierarchy_sec_snippets = []
    elif not strict_uploaded_only:
        # Prepend the firm standard paragraph so the model sees approved wording
        # first. In hybrid mode this applies at EVERY tier (medium/maximum too),
        # not only the ≤12% assembly tier — otherwise the firm voice never
        # reaches higher-involvement drafts (the boss's "ignored our paragraphs"
        # complaint).
        std_tier_ok = ap_inv <= 12 or shape_with_std
        if section_std_prepared and std_tier_ok:
            para_snippets = [section_std_prepared] + [
                p for p in para_snippets if p.strip() != section_std_prepared
            ]

    # Hybrid directive: the firm standard paragraphs are a menu of variant
    # sentences with placeholders. Tell the model how to consume them so it does
    # not emit contradictory variants or literal placeholders/option brackets.
    if shape_with_std and section_std_prepared and ap_inv > 0:
        std_directive = (
            " STANDARD-PARAGRAPH USE: the STANDARD SOURCE PASSAGES are the firm's "
            "approved wording for this section and contain ALTERNATIVE variant "
            "sentences plus placeholders. Adopt the firm's phrasing and structure. "
            "SELECT only the variant(s) consistent with the RAW NOTES and DISCARD "
            "the others (never include mutually contradictory variants). Replace "
            "every '<text>' placeholder and choose between '(option)(option)' "
            "brackets using the RAW NOTES; if the note does not specify, omit that "
            "clause. NEVER output a literal '<text>', angle brackets, or unresolved "
            "'( )' option brackets."
        )
        level_params["creativity_hint"] = (
            str(level_params.get("creativity_hint") or "") + std_directive
        )

    seen_txt: set[str] = set()
    verify_snippets: list[str] = []
    for t in doc_snippets + hierarchy_sec_snippets + para_snippets:
        if t in seen_txt:
            continue
        seen_txt.add(t)
        verify_snippets.append(t)

    # Low involvement: suppress style-matching prompts so the model does not
    # paraphrase toward a learned voice; retrieved / standard wording dominates.
    effective_profile = None if int(level_params.get("ai_percent", 50)) <= 20 else style_profile

    # ── Style-corpus paragraph injection ───────────────────────────────────
    # When the tenant has uploaded PAST reports (purpose=style_corpus), we
    # pull a few topically relevant verbatim paragraphs and merge them into
    # the profile's example_paragraphs for THIS call. These power the LLM's
    # voice mirroring without entering the factual evidence pool
    # (verify_snippets is unchanged) so observations from old jobs cannot
    # leak into the new report.
    if effective_profile is not None and bullets:
        try:
            from app.services.style_corpus import retrieve_style_paragraphs

            style_query = (template.title if template else template_id)
            if isinstance(bullets, list) and bullets:
                style_query = f"{style_query} {bullets[0]}"
            style_hits = await retrieve_style_paragraphs(
                tenant_id=tenant_id, query=style_query, k=4
            )
            extra_examples = [
                (r.text or "").strip() for r in style_hits if (r.text or "").strip()
            ]
            if extra_examples:
                # De-duplicate against existing examples, cap total at 5 so
                # the prompt budget stays bounded.
                merged = list(effective_profile.example_paragraphs or [])
                merged_lower = {e.casefold() for e in merged}
                for ex in extra_examples:
                    if ex.casefold() in merged_lower:
                        continue
                    merged.append(ex)
                    if len(merged) >= 5:
                        break
                effective_profile = WritingStyleProfile(
                    **{**effective_profile.model_dump(), "example_paragraphs": merged}
                )
                logger.debug(
                    "Style corpus injected %d paragraph(s) for tenant=%s section=%s",
                    len(extra_examples),
                    tenant_id,
                    template_id,
                )
        except Exception as exc:  # noqa: BLE001
            logger.debug(
                "Style-corpus paragraph injection skipped tenant=%s section=%s: %s",
                tenant_id,
                template_id,
                exc,
            )

    identity = _extract_property_identity(list(bullets))
    identity_block = _identity_facts_block(identity)

    # Structural-router path at ai_percent == 0: use the LLM as a constrained
    # router that takes RAG-retrieved STANDARD passages and weaves the
    # inspector's NOTES specifics into the appropriate slots — no creative
    # writing, no new sentences, no paraphrasing of standard wording.
    # If the LLM is unavailable (mock/no key) or returns empty, fall back to
    # the deterministic stitcher which guarantees an output without an LLM call.
    raw_text: str = ""
    ap_branch = int(level_params.get("ai_percent", 50))
    if ap_branch == 0:
        # Combine all retrieved standards in the order the deterministic
        # stitcher would: paragraph > section > document.
        standards = [
            *(s for s in (para_snippets or []) if s),
            *(s for s in (hierarchy_sec_snippets or []) if s),
            *(s for s in (doc_snippets or []) if s),
        ]
        section_title = template.title if template else None
        try:
            raw_text = await gen_llm.constrained_weave(
                section_code=template_id,
                section_title=section_title,
                bullets=list(expanded_bullets),
                standard_passages=standards,
                survey_level=report_survey_level,
                tenant_id=tenant_id,
            )
        except Exception as exc:  # noqa: BLE001
            logger.warning("constrained_weave call failed: %s", exc)
            raw_text = ""

        if raw_text:
            logger.info(
                "Constrained-weave router used for section=%s (ai_percent=0): %d chars",
                template_id,
                len(raw_text),
            )
        else:
            # Deterministic fallback — splice retrieved standard wording with
            # raw notes appended, no LLM dependency.
            raw_text = _deterministic_assembly_stitch(
                skeleton=skeleton,
                bullets=list(expanded_bullets),
                paragraph_snippets=para_snippets,
                document_snippets=doc_snippets,
                hierarchy_section_snippets=hierarchy_sec_snippets,
                survey_level=report_survey_level,
                redact_references=notes_only,
            )
            if raw_text:
                logger.info(
                    "Deterministic stitcher fallback for section=%s (ai_percent=0): %d chars",
                    template_id,
                    len(raw_text),
                )
            else:
                logger.info(
                    "No standard paragraph or weave for section=%s — returning blank for 0%% tier",
                    template_id,
                )
                # 0% contract: do not generate prose from other sources.
                zero_metrics: dict[str, Any] = {"requested_ai_percent": int(ap_branch)}
                if clamp_report.has_clamps:
                    zero_metrics["bullet_clamps"] = clamp_report.to_dict()
                return "", [], 0.0, [], [], zero_metrics

    if not raw_text:
        raw_text = await gen_llm.generate_section(
            skeleton=skeleton,
            bullets=expanded_bullets,
            snippets=[],
            style_profile=effective_profile,
            temperature=level_params["temperature"],
            creativity_hint=level_params["creativity_hint"],
            document_context=doc_snippets,
            style_anchor=draft_paragraph,
            hierarchy_section_snippets=hierarchy_sec_snippets if hierarchy_sec_snippets else None,
            paragraph_snippets=para_snippets,
            identity_facts=identity_block,
            survey_level=report_survey_level,
            reference_only_context=notes_only,
            ai_percent=ap_branch,
            interference_level=interference_level,
            tenant_id=tenant_id,
        )
    # Remove drafting scaffold sub-headings ("Executive Summary", "Defects and
    # Risks", etc.) that the model echoes into the prose, BEFORE grounding —
    # otherwise the non-invention pass mangles them into fragments and clips the
    # leading word of the following sentence.
    raw_text = strip_scaffold_subheadings(raw_text)
    # Strip the leaked "<CODE> — <Title>" heading the model echoes at the start
    # of the body (the section header is rendered separately by the exporter).
    _sec_title = (template.title if template else "").strip()
    _lead_heading = re.compile(
        rf"^\s*{re.escape(template_id)}\s*[—–-]\s*"
        + (rf"(?:{re.escape(_sec_title)})?" if _sec_title else "")
        + r"[ \t]*\n?",
        re.IGNORECASE,
    )
    raw_text = _lead_heading.sub("", raw_text, count=1).lstrip()
    # Layer 1+2 grounding: regex fast pass + LLM context-aware grounding.
    # Photo observations are included as trusted evidence so vision-derived
    # findings survive the grounding pass instead of being stripped.
    text = await async_enforce_verify(
        text=raw_text,
        bullets=grounding_evidence,
        snippets=[] if notes_only else verify_snippets,
        pinned_identity=identity,
        openai_api_key=settings.openai_api_key,
        model=settings.chat_model,
    )

    # Verbatim-ratio enforcement: the slider value is a contract.
    # If the user picks 25%, ~75% of the output should be verbatim from the
    # standard sources. If the model drifts substantially below the floor
    # (because it ignored the prompt and paraphrased), regenerate once with
    # a stricter hint that quotes the actual measured shortfall back to it.
    target_ratio, floor_ratio = verbatim_ratio_target(ap_branch)
    overlap = verbatim_overlap_ratio(text, verify_snippets, n=6) if verify_snippets else 0.0
    if floor_ratio > 0.0 and verify_snippets and settings.openai_api_key and ap_branch > 0:
        best_text = text
        best_overlap = overlap
        for attempt in (1, 2):
            if best_overlap >= floor_ratio:
                break
            logger.warning(
                "Verbatim ratio miss for section=%s ai_percent=%d: measured=%.2f floor=%.2f target=%.2f — retry %d/2",
                template_id,
                ap_branch,
                best_overlap,
                floor_ratio,
                target_ratio,
                attempt,
            )
            ratio_hint = (
                f"AI INVOLVEMENT CONTRACT VIOLATION. The user set the slider to {ap_branch}%. "
                f"That requires roughly {int(round(target_ratio * 100))}% of the wording to be VERBATIM "
                f"from the STANDARD SOURCE PASSAGES, the SECTION SKELETON, or the RAW NOTES. "
                f"Your previous draft was only {int(round(best_overlap * 100))}% verbatim. "
                "Regenerate now: copy applicable sentences word-for-word from the SOURCE PASSAGES. "
                "DO NOT replace any source word with a synonym. New tokens permitted: short connectors only."
            )
            retry_raw = await gen_llm.generate_section(
                skeleton=skeleton,
                bullets=expanded_bullets,
                snippets=[],
                style_profile=effective_profile,
                temperature=0.0,
                creativity_hint=ratio_hint,
                document_context=doc_snippets,
                style_anchor=draft_paragraph,
                hierarchy_section_snippets=hierarchy_sec_snippets if hierarchy_sec_snippets else None,
                paragraph_snippets=para_snippets,
                identity_facts=identity_block,
                survey_level=report_survey_level,
                reference_only_context=notes_only,
                ai_percent=ap_branch,
                interference_level=interference_level,
                tenant_id=tenant_id,
            )
            retry_text = await async_enforce_verify(
                text=retry_raw,
                bullets=grounding_evidence,
                snippets=[] if notes_only else verify_snippets,
                pinned_identity=identity,
                openai_api_key=settings.openai_api_key,
                model=settings.chat_model,
            )
            retry_overlap = verbatim_overlap_ratio(retry_text, verify_snippets, n=6)
            if retry_overlap >= best_overlap:
                best_text = retry_text
                best_overlap = retry_overlap
        text = best_text
        overlap = best_overlap
    else:
        logger.debug(
            "Verbatim overlap at section=%s ai_percent=%d: measured=%.2f floor=%.2f target=%.2f",
            template_id,
            ap_branch,
            overlap,
            floor_ratio,
            target_ratio,
        )

    measured_ai_percent = int(round(100.0 * max(0.0, min(1.0, 1.0 - overlap))))
    metrics: dict[str, Any] = {
        "requested_ai_percent": int(ap_branch),
        "measured_ai_percent": measured_ai_percent,
        "verbatim_overlap": float(round(overlap, 4)),
        "target_verbatim_percent": int(round(target_ratio * 100)),
        "floor_verbatim_percent": int(round(floor_ratio * 100)),
        "verbatim_ngram_n": 6,
        "photo_observation_count": len(photo_obs or []),
        "photo_evidence_used": bool(photo_obs),
    }
    if photo_note:
        metrics["photo_note"] = photo_note
    if clamp_report.has_clamps:
        # Surface the clamp warning to the API caller — without this the user
        # only sees a server log line and cannot tell the report is missing
        # data. Used by the UI/log to render "N inputs dropped from this section".
        metrics["bullet_clamps"] = clamp_report.to_dict()

    # Identity guard: if we detect address/type/occupancy contradictions, re-run once with a stricter hint/temperature.
    issues = _detect_identity_contradictions(text, identity)
    if issues and settings.openai_api_key:
        strict_hint = (
            "CRITICAL CONSISTENCY FIX: The previous draft contradicted the pinned PROPERTY IDENTITY. "
            "You MUST remove any other addresses/postcodes and any flats/communal language unless explicitly supported. "
            "If information is missing or cannot be verified, omit the unsupported claim; do not use placeholder "
            "sentences. "
            "Issues detected: " + "; ".join(issues)
        )
        retry_raw = await gen_llm.generate_section(
            skeleton=skeleton,
            bullets=expanded_bullets,
            snippets=[],
            style_profile=effective_profile,
            temperature=min(0.12, float(level_params["temperature"])),
            creativity_hint=strict_hint,
            document_context=doc_snippets,
            style_anchor=draft_paragraph,
            hierarchy_section_snippets=hierarchy_sec_snippets if hierarchy_sec_snippets else None,
            paragraph_snippets=para_snippets,
            identity_facts=identity_block,
            survey_level=report_survey_level,
            reference_only_context=notes_only,
            ai_percent=int(level_params.get("ai_percent", 50)),
            interference_level=interference_level,
        )
        text = await async_enforce_verify(
            text=retry_raw,
            bullets=bullets,
            snippets=[] if notes_only else verify_snippets,
            pinned_identity=identity,
            openai_api_key=settings.openai_api_key,
            model=settings.chat_model,
        )
        issues = _detect_identity_contradictions(text, identity)
        if issues:
            logger.warning(
                "Identity contradictions persist after strict retry (section evidence may be thin): %s",
                "; ".join(issues),
            )

    # Survey-level behavioural enforcement: one retry with a strict hint.
    tier_issues = _tier_validation_issues(text, report_survey_level)
    if tier_issues and settings.openai_api_key:
        tier_hint = (
            "CRITICAL SURVEY LEVEL COMPLIANCE FIX: The previous draft did not comply with the required "
            "RICS survey level behaviour. You MUST correct this now.\n"
            "If information is missing or cannot be verified from notes/evidence, omit the unsupported claim "
            "instead of writing placeholder sentences.\n"
            "Issues detected: " + "; ".join(tier_issues)
        )
        retry_raw = await gen_llm.generate_section(
            skeleton=skeleton,
            bullets=expanded_bullets,
            snippets=[],
            style_profile=effective_profile,
            temperature=min(0.15, float(level_params["temperature"])),
            creativity_hint=tier_hint,
            document_context=doc_snippets,
            style_anchor=draft_paragraph,
            hierarchy_section_snippets=hierarchy_sec_snippets if hierarchy_sec_snippets else None,
            paragraph_snippets=para_snippets,
            identity_facts=identity_block,
            survey_level=report_survey_level,
            reference_only_context=notes_only,
            ai_percent=int(level_params.get("ai_percent", 50)),
            interference_level=interference_level,
        )
        text = await async_enforce_verify(
            text=retry_raw,
            bullets=bullets,
            snippets=[] if notes_only else verify_snippets,
            pinned_identity=identity,
            openai_api_key=settings.openai_api_key,
            model=settings.chat_model,
        )
        tier_issues = _tier_validation_issues(text, report_survey_level)
        if tier_issues:
            logger.warning("Tier validation issues persist after retry: %s", "; ".join(tier_issues))

    # ── RAW NOTES COVERAGE RETRY (up to 3 attempts) ───────────────────────
    # Measure how much of the inspector's bullets actually surfaced in the
    # generated text. If coverage drops below the threshold (≥70% of fact
    # tokens, ≤3 poorly covered bullets) we regenerate with an explicit
    # hint listing the dropped observations. We keep the best attempt by
    # coverage ratio across all retries so a worse second attempt cannot
    # replace a better first one.
    from app.generator.notes_coverage import (
        coverage_regenerate_hint,
        coverage_report,
    )

    best_text = text
    best_coverage = coverage_report(list(bullets), best_text)
    metrics["coverage_ratio"] = round(best_coverage.coverage_ratio, 4)
    metrics["coverage_bullets_total"] = best_coverage.bullets_total
    metrics["coverage_bullets_poor"] = best_coverage.bullets_poorly_covered
    metrics["coverage_attempts"] = 0

    # ``text`` here was already run through the full LLM grounding pass. During
    # retries we score candidates with the cheap *regex-only* grounding pass and
    # run the expensive LLM grounding once, on the winning draft — avoiding one
    # LLM round-trip per retry (the single biggest per-section latency saving
    # when coverage is poor).
    max_coverage_retries = int(getattr(settings, "max_coverage_retries", 3))
    coverage_retry_won = False
    if settings.openai_api_key and bullets:
        for attempt in range(1, max_coverage_retries + 1):
            if not best_coverage.needs_regenerate:
                break
            hint = coverage_regenerate_hint(best_coverage)
            if not hint:
                break
            logger.info(
                "Notes-coverage retry %d/%d for section=%s: coverage=%.2f poor_bullets=%d",
                attempt,
                max_coverage_retries,
                template_id,
                best_coverage.coverage_ratio,
                best_coverage.bullets_poorly_covered,
            )
            retry_raw = await gen_llm.generate_section(
                skeleton=skeleton,
                bullets=expanded_bullets,
                snippets=[],
                style_profile=effective_profile,
                temperature=min(0.18, float(level_params["temperature"])),
                creativity_hint=hint,
                document_context=doc_snippets,
                style_anchor=draft_paragraph,
                hierarchy_section_snippets=hierarchy_sec_snippets if hierarchy_sec_snippets else None,
                paragraph_snippets=para_snippets,
                identity_facts=identity_block,
                survey_level=report_survey_level,
                reference_only_context=notes_only,
                ai_percent=int(level_params.get("ai_percent", 50)),
                interference_level=interference_level,
                tenant_id=tenant_id,
            )
            retry_text = enforce_verify(
                text=retry_raw,
                bullets=bullets,
                snippets=[] if notes_only else verify_snippets,
                pinned_identity=identity,
            )
            retry_coverage = coverage_report(list(bullets), retry_text)
            metrics["coverage_attempts"] = attempt
            if retry_coverage.coverage_ratio > best_coverage.coverage_ratio:
                best_text = retry_text
                best_coverage = retry_coverage
                coverage_retry_won = True
                metrics["coverage_ratio"] = round(best_coverage.coverage_ratio, 4)
                metrics["coverage_bullets_total"] = best_coverage.bullets_total
                metrics["coverage_bullets_poor"] = best_coverage.bullets_poorly_covered
        if coverage_retry_won:
            # Winning draft only saw the regex pass above; apply the full LLM
            # grounding once so the persisted text keeps the same guarantee as
            # the non-retry path.
            best_text = await async_enforce_verify(
                text=best_text,
                bullets=bullets,
                snippets=[] if notes_only else verify_snippets,
                pinned_identity=identity,
                openai_api_key=settings.openai_api_key,
                model=settings.chat_model,
            )
    text = best_text
    if best_coverage.bullets_total and best_coverage.needs_regenerate:
        # Surface remaining gaps so the UI/log can flag them without blocking.
        metrics["coverage_missing_tokens"] = list(best_coverage.missing_tokens[:32])

    # Optional LLM validator loop: PASS/FAIL with reasons, then regenerate once with feedback.
    if settings.llm_section_validator_enabled and settings.openai_api_key:
        max_retries = int(getattr(settings, "llm_section_validator_max_retries", 1) or 0)
        attempts = 0
        while attempts < max_retries:
            verdict_raw = await gen_llm.validate_section_compliance(
                survey_level=report_survey_level,
                section_code=template_id,
                bullets=bullets,
                evidence_snippets=verify_snippets,
                text=text,
            )
            ok, verdict = _parse_validator_result(verdict_raw)
            if ok:
                break
            attempts += 1
            fb_hint = (
                "COMPLIANCE VALIDATION FAILED. You MUST regenerate the paragraph so it passes the validator.\n"
                "Validator output:\n"
                f"{verdict}\n\n"
                "Remember: do not invent facts; only use RAW NOTES and evidence. "
                "If missing, omit the unsupported claim."
            )
            retry_raw = await gen_llm.generate_section(
                skeleton=skeleton,
                bullets=expanded_bullets,
                snippets=[],
                style_profile=effective_profile,
                temperature=min(0.18, float(level_params["temperature"])),
                creativity_hint=fb_hint,
                document_context=doc_snippets,
                style_anchor=draft_paragraph,
                hierarchy_section_snippets=hierarchy_sec_snippets if hierarchy_sec_snippets else None,
                paragraph_snippets=para_snippets,
                identity_facts=identity_block,
                survey_level=report_survey_level,
                reference_only_context=notes_only,
                ai_percent=int(level_params.get("ai_percent", 50)),
                interference_level=interference_level,
                tenant_id=tenant_id,
            )
            text = await async_enforce_verify(
                text=retry_raw,
                bullets=bullets,
                snippets=[] if notes_only else verify_snippets,
                pinned_identity=identity,
                openai_api_key=settings.openai_api_key,
                model=settings.chat_model,
            )
        else:
            logger.warning("LLM validator retries exhausted for section=%s", template_id)

    prov_rows: list[dict[str, Any]] = []
    if notes_only:
        # Hard guarantee: do not surface provenance citations from other docs.
        provenance = []
        confidence = 0.0
        return text, provenance, confidence, [], [], metrics
    seen_chunks: set[str] = set()
    for r in doc_ctx_results + list(top_results):
        if r.chunk_id in seen_chunks:
            continue
        seen_chunks.add(r.chunk_id)
        prov_rows.append(
            Provenance(doc_id=r.doc_id, chunk_id=r.chunk_id, score=round(r.score, 4)).model_dump()
        )
    provenance = prov_rows
    confidence = (
        sum(r.rerank_score for r in top_results) / len(top_results) if top_results else 0.0
    )
    return text, provenance, confidence, top_results, doc_ctx_results, metrics


async def _attach_citation_audit(
    *,
    metrics: dict[str, Any],
    template_id: str,
    survey_level: int | None,
    evidence: list[SearchResult],
    tenant_id: str,
) -> None:
    """Run the deterministic grounding audit and record it in ``metrics``.

    Best-effort and fully isolated: any failure here must never affect the
    generated section, so the audit is wrapped and only logged. This is an
    additive traceability artifact, not part of the generation contract.
    """
    try:
        from app.extraction.extractor import audit_section_grounding
        from app.templates.registry import get_template

        tmpl = get_template(template_id, survey_level)
        section_name = tmpl.title if tmpl else template_id

        extraction = await audit_section_grounding(
            section_name=section_name,
            template_id=template_id,
            chunks=evidence,
            tenant_id=tenant_id,
        )
        metrics["citation_audit"] = {
            "section": extraction.section,
            "confidence": extraction.confidence,
            "findings": len(extraction.findings),
            "contradictions": [c.model_dump() for c in extraction.contradictions],
            "dropped_claims": extraction.dropped_claims,
        }
        if extraction.contradictions or extraction.dropped_claims:
            logger.warning(
                "Citation audit section=%s confidence=%.2f findings=%d "
                "contradictions=%d dropped=%d",
                extraction.section,
                extraction.confidence,
                len(extraction.findings),
                len(extraction.contradictions),
                len(extraction.dropped_claims),
            )
        else:
            logger.info(
                "Citation audit section=%s confidence=%.2f findings=%d (clean)",
                extraction.section,
                extraction.confidence,
                len(extraction.findings),
            )
    except Exception as exc:  # noqa: BLE001 — audit must never break generation
        logger.warning("Citation audit skipped for section=%s: %s", template_id, exc)


# ── GENERATE mode ──────────────────────────────────────────────────────────────

async def _run_generate(
    db: AsyncSession,
    report: Report,
    tenant_id: str,
    template_id: str,
    bullets: list[str],
    force_regenerate: bool,
    ai_level: int = AILevel.balanced,
    ai_percent: int | None = None,
    retrieval_level: str = "paragraph",
    reference_document_ids: list[str] | None = None,
    draft_paragraph: str | None = None,
    pipeline: str = "standard",
    fallback_used: bool = False,
    strict_uploaded_only: bool = False,
    interference_level: str | None = None,
    mark_report_complete: bool = True,
) -> None:
    """Style-aware RAG generation pipeline."""
    notes_only = bool(getattr(settings, "notes_only_generation", True))
    # In notes-only mode, reference docs are still allowed for *understanding*
    # (structure/terminology), but provenance is suppressed and facts must come from notes.
    refs = list(reference_document_ids or [])
    ai_norm = _ai_level_to_params(ai_level, ai_percent=ai_percent)
    # 1. Cache check (ai_level is part of the key: level-1 ≠ level-5 results)
    cache_key = section_cache.compute_cache_key(
        template_id,
        bullets,
        tenant_id,
        ai_level=ai_level,
        ai_percent=ai_percent,
        reference_document_ids=refs,
        draft_paragraph=draft_paragraph,
        rics_survey_level=report.survey_level,
        interference_level=interference_level,
    )
    if not force_regenerate:
        cached = section_cache.get(cache_key)
        if cached:
            logger.info("Cache hit for report=%s section=%s", report.id, template_id)
            cached_sp_dict = cached.get("style_profile")
            cached_style_profile = (
                WritingStyleProfile(**cached_sp_dict)
                if isinstance(cached_sp_dict, dict)
                else await _get_or_build_style_profile(tenant_id)
            )
            prov = cached.get("provenance", [])
            _tier_cached = _interference_from_cache_entry(cached)
            await _persist_section(
                db=db,
                report=report,
                section_code=template_id,
                text=cached["text"],
                confidence=cached.get("confidence", 0.0),
                provenance=prov,
                cached=True,
                mode=GenerationMode.generate,
                style_profile=cached_style_profile,
                ai_level=int(cached.get("ai_level", ai_level)),
                ai_percent=int(cached.get("ai_percent")) if cached.get("ai_percent") is not None else ai_percent,
                measured_ai_percent=int(cached.get("measured_ai_percent")) if cached.get("measured_ai_percent") is not None else None,
                verbatim_overlap=float(cached.get("verbatim_overlap")) if cached.get("verbatim_overlap") is not None else None,
                pipeline=str(cached.get("pipeline") or pipeline),
                fallback_used=bool(cached.get("fallback_used", fallback_used)),
                interference_level=_tier_cached or interference_level,
                mark_report_complete=mark_report_complete,
                citation_audit=cached.get("citation_audit") if isinstance(cached.get("citation_audit"), dict) else None,
            )
            return

    # 2. Style profile (tenant-wide). In strict_uploaded_only mode we do not
    # look at any tenant library history; we keep style neutral so only the
    # report's attached docs + bullets influence output.
    style_profile = (
        await _get_or_build_style_profile(tenant_id)
        if not strict_uploaded_only
        else WritingStyleProfile()
    )
    if notes_only and style_profile:
        # Prevent verbatim style "example_paragraphs" from leaking tenant-library
        # content into prompts. Keep tone/phrases/patterns only.
        try:
            style_profile = WritingStyleProfile(**{**style_profile.model_dump(), "example_paragraphs": []})
        except Exception:  # noqa: BLE001
            pass

    # 3–7. Generate text (no DB writes inside this call)
    text, provenance, confidence, top_results, doc_ctx_results, metrics = await _generate_section_text(
        tenant_id=tenant_id,
        template_id=template_id,
        bullets=bullets,
        style_profile=style_profile,
        ai_level=ai_level,
        ai_percent=ai_percent,
        primary_document_id=report.document_id,
        reference_document_ids=refs,
        draft_paragraph=draft_paragraph,
        report_id=str(report.id),
        retrieval_level=retrieval_level,
        db=db,
        report_survey_level=report.survey_level,
        strict_uploaded_only=strict_uploaded_only,
        interference_level=interference_level,
    )
    doc_ids = {str(p.get("doc_id", "")) for p in provenance if p.get("doc_id")}
    filenames = await fetch_doc_filenames(db, tenant_id, doc_ids)
    merged_for_meta: list[SearchResult] = []
    seen_meta: set[str] = set()
    for r in doc_ctx_results + list(top_results):
        if r.chunk_id in seen_meta:
            continue
        seen_meta.add(r.chunk_id)
        merged_for_meta.append(r)
    provenance = attach_snippet_metadata(provenance, merged_for_meta, filenames)

    # 7b. Optional citation-grounded audit (non-destructive). Runs the
    # deterministic extraction + contradiction layer over the SAME retrieved
    # evidence the section was built from, scoped to this section's domain.
    # It never mutates `text`; it only records traceability (confidence,
    # contradictions, dropped/ungrounded claims) into metrics for auditing.
    if settings.enable_citation_extraction and top_results:
        await _attach_citation_audit(
            metrics=metrics,
            template_id=template_id,
            survey_level=report.survey_level,
            evidence=list(top_results),
            tenant_id=tenant_id,
        )

    # 8. Cache & persist (ai_level stored for auditability; it is already part of the key)
    section_cache.set(cache_key, {
        "text": text,
        "confidence": confidence,
        "provenance": provenance,
        "style_profile": json.loads(style_profile.model_dump_json()),
        "ai_level": int(ai_norm.get("ai_level", ai_level)),
        "ai_percent": int(ai_norm.get("ai_percent", ai_level_to_percent(int(ai_level)))),
        "measured_ai_percent": metrics.get("measured_ai_percent"),
        "verbatim_overlap": metrics.get("verbatim_overlap"),
        "pipeline": str(pipeline),
        "fallback_used": bool(fallback_used),
        "interference_level": interference_level,
        "citation_audit": metrics.get("citation_audit") if isinstance(metrics.get("citation_audit"), dict) else None,
    })
    _bullet_clamps_meta = (
        metrics.get("bullet_clamps")
        if isinstance(metrics.get("bullet_clamps"), dict)
        else None
    )
    _coverage_meta: dict[str, Any] | None = None
    if "coverage_ratio" in metrics:
        _coverage_meta = {
            "coverage_ratio": metrics.get("coverage_ratio"),
            "bullets_total": metrics.get("coverage_bullets_total"),
            "bullets_poorly_covered": metrics.get("coverage_bullets_poor"),
            "attempts": metrics.get("coverage_attempts", 0),
        }
        miss = metrics.get("coverage_missing_tokens")
        if miss:
            _coverage_meta["missing_tokens"] = list(miss)
    await _persist_section(
        db=db,
        report=report,
        section_code=template_id,
        text=text,
        confidence=confidence,
        provenance=provenance,
        cached=False,
        mode=GenerationMode.generate,
        style_profile=style_profile,
        ai_level=int(ai_norm.get("ai_level", ai_level)),
        ai_percent=int(ai_norm.get("ai_percent")) if ai_norm.get("ai_percent") is not None else None,
        measured_ai_percent=int(metrics["measured_ai_percent"]) if isinstance(metrics.get("measured_ai_percent"), int) else None,
        verbatim_overlap=float(metrics["verbatim_overlap"]) if isinstance(metrics.get("verbatim_overlap"), float) else None,
        pipeline=str(pipeline),
        fallback_used=bool(fallback_used),
        interference_level=interference_level,
        mark_report_complete=mark_report_complete,
        bullet_clamps=_bullet_clamps_meta,
        coverage_metrics=_coverage_meta,
        citation_audit=metrics.get("citation_audit") if isinstance(metrics.get("citation_audit"), dict) else None,
    )
    logger.info(
        "Generated section=%s for report=%s (confidence=%.3f, style=%s, ai=%s%% level=%s)",
        template_id,
        report.id,
        confidence,
        style_profile.tone,
        int(ai_norm.get("ai_percent", 50)),
        int(ai_norm.get("ai_level", ai_level)),
    )


def _pipeline_setting() -> str:
    raw = str(
        getattr(settings, "primary_generate_pipeline", "standard") or ""
    ).strip().lower()
    return raw if raw in ("agentic", "standard") else "standard"


async def _run_generate_primary(
    *,
    db: AsyncSession,
    report: Report,
    tenant_id: str,
    template_id: str,
    bullets: list[str],
    force_regenerate: bool,
    ai_level: int = AILevel.balanced,
    ai_percent: int | None = None,
    retrieval_level: str = "paragraph",
    reference_document_ids: list[str] | None = None,
    draft_paragraph: str | None = None,
    strict_uploaded_only: bool = False,
    interference_level: str | None = None,
    mark_report_complete: bool = True,
) -> None:
    """Primary generate dispatcher: agentic-first (with fallback) or standard-only."""
    notes_only = bool(getattr(settings, "notes_only_generation", True))
    allow_agentic_notes_only = bool(
        getattr(settings, "agentic_inspector_when_notes_only", False)
    )
    if notes_only and not allow_agentic_notes_only:
        # notes_only: force standard generator (avoid agentic tool-loop), but allow
        # reference docs for *understanding* (they are redacted and never cited).
        await _run_generate(
            db=db,
            report=report,
            tenant_id=tenant_id,
            template_id=template_id,
            bullets=bullets,
            force_regenerate=force_regenerate,
            ai_level=ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            reference_document_ids=reference_document_ids,
            draft_paragraph=draft_paragraph,
            pipeline="standard",
            fallback_used=False,
            strict_uploaded_only=strict_uploaded_only,
            interference_level=interference_level,
            mark_report_complete=mark_report_complete,
        )
        return
    primary = _pipeline_setting()
    if primary == "standard":
        await _run_generate(
            db=db,
            report=report,
            tenant_id=tenant_id,
            template_id=template_id,
            bullets=bullets,
            force_regenerate=force_regenerate,
            ai_level=ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            reference_document_ids=reference_document_ids,
            draft_paragraph=draft_paragraph,
            pipeline="standard",
            fallback_used=False,
            strict_uploaded_only=strict_uploaded_only,
            interference_level=interference_level,
            mark_report_complete=mark_report_complete,
        )
        return

    # Agentic-first: try HeadAgent (tool-calling when live), fall back to standard on any error.
    try:
        await _run_generate_agentic(
            db=db,
            report=report,
            tenant_id=tenant_id,
            template_id=template_id,
            bullets=bullets,
            force_regenerate=force_regenerate,
            ai_level=ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            reference_document_ids=reference_document_ids,
            draft_paragraph=draft_paragraph,
            interference_level=interference_level,
            mark_report_complete=mark_report_complete,
        )
    except Exception as exc:  # noqa: BLE001
        logger.exception(
            "Agentic generate failed; falling back to standard. report=%s section=%s err=%s",
            report.id,
            template_id,
            exc,
        )
        await _run_generate(
            db=db,
            report=report,
            tenant_id=tenant_id,
            template_id=template_id,
            bullets=bullets,
            force_regenerate=force_regenerate,
            ai_level=ai_level,
            ai_percent=ai_percent,
            retrieval_level=retrieval_level,
            reference_document_ids=reference_document_ids,
            draft_paragraph=draft_paragraph,
            pipeline="standard",
            fallback_used=True,
            strict_uploaded_only=strict_uploaded_only,
            interference_level=interference_level,
            mark_report_complete=mark_report_complete,
        )


async def _run_generate_agentic(
    *,
    db: AsyncSession,
    report: Report,
    tenant_id: str,
    template_id: str,
    bullets: list[str],
    force_regenerate: bool,
    ai_level: int = AILevel.balanced,
    ai_percent: int | None = None,
    retrieval_level: str = "paragraph",
    reference_document_ids: list[str] | None = None,
    draft_paragraph: str | None = None,
    interference_level: str | None = None,
    mark_report_complete: bool = True,
) -> None:
    """Agentic per-section generate (HeadAgent) with the same cache key as standard generate."""
    # Clamp + dedupe upfront so the cache key, the LLM input, and the
    # clamp-warning metric all agree on what bullets were actually used.
    # The standard path does the same thing inside `_generate_section_text`;
    # the agentic path was previously silently clamping inside the bullets-
    # to-prompt formatter — the user couldn't see what was dropped.
    from app.generator.note_bullets import clean_and_clamp_bullets_with_report

    bullet_cap = int(settings.max_bullets_per_section)
    bullets, agentic_clamp_report = clean_and_clamp_bullets_with_report(
        list(bullets or []), max_items=bullet_cap
    )
    if agentic_clamp_report.has_clamps:
        logger.warning(
            "Agentic section %s notes clamped: input=%d kept=%d dropped=%d (cap=%d)",
            template_id,
            agentic_clamp_report.input_count,
            agentic_clamp_report.cleaned_count,
            agentic_clamp_report.total_dropped,
            bullet_cap,
        )
    # Keep the same cache key behavior so switching pipelines doesn't silently ignore cache controls.
    refs = list(reference_document_ids or [])
    ai_norm = _ai_level_to_params(ai_level, ai_percent=ai_percent)
    cache_key = section_cache.compute_cache_key(
        template_id,
        bullets,
        tenant_id,
        ai_level=ai_level,
        ai_percent=ai_percent,
        reference_document_ids=refs,
        draft_paragraph=draft_paragraph,
        rics_survey_level=report.survey_level,
        interference_level=interference_level,
    )
    if not force_regenerate:
        cached = section_cache.get(cache_key)
        if cached:
            cached_sp_dict = cached.get("style_profile")
            cached_style_profile = (
                WritingStyleProfile(**cached_sp_dict)
                if isinstance(cached_sp_dict, dict)
                else await _get_or_build_style_profile(tenant_id)
            )
            prov = cached.get("provenance", [])
            cached_insp = cached.get("inspector")
            inspector_cached = cached_insp if isinstance(cached_insp, dict) else None
            await _persist_section(
                db=db,
                report=report,
                section_code=template_id,
                text=cached["text"],
                confidence=cached.get("confidence", 0.0),
                provenance=prov,
                cached=True,
                mode=GenerationMode.generate,
                style_profile=cached_style_profile,
                ai_level=int(cached.get("ai_level", ai_level)),
                ai_percent=int(cached.get("ai_percent")) if cached.get("ai_percent") is not None else ai_percent,
                pipeline="agentic",
                fallback_used=bool(cached.get("fallback_used", False)),
                inspector=inspector_cached,
                interference_level=_interference_from_cache_entry(cached) or interference_level,
                mark_report_complete=mark_report_complete,
            )
            return

    style_profile = await _get_or_build_style_profile(tenant_id)

    # Use the agentic HeadAgent per-section generator.
    from app.agentic.agents import HeadAgent, render_report_text

    head = HeadAgent()
    rep = await head.generate_section_report(
        db=db,
        tenant_id=tenant_id,
        primary_document_id=report.document_id,
        section_code=template_id,
        bullets=bullets,
        style_profile=style_profile,
        ai_percent=int(ai_norm.get("ai_percent", 50)),
        retrieval_level=retrieval_level,
        reference_document_ids=refs,
        similarity_scan=False,
        peer_sections=None,
        similarity_exclude_document_ids=None,
        survey_level=report.survey_level,
    )
    inspector_meta = _agentic_inspector_meta(rep)

    # Convert structured blocks to a single section text string.
    text = render_report_text(
        title=f"RICS Inspection Report — Section {template_id}",
        blocks={
            "Executive Summary": rep.executive_summary,
            "Property Description": rep.property_description,
            "Condition Assessment": rep.condition_assessment,
            "Defects and Risks": rep.defects_and_risks,
            "Recommendations": rep.recommendations,
        },
        section_code=template_id,
        survey_level=report.survey_level,
    )

    # Build provenance from evidence items, then attach filenames/snippet previews like standard.
    prov_rows: list[dict[str, Any]] = []
    for e in rep.evidence_items:
        if not e.doc_id or not e.chunk_id:
            continue
        prov_rows.append(
            Provenance(
                doc_id=str(e.doc_id),
                chunk_id=str(e.chunk_id),
                score=round(float(e.score), 4),
                snippet_preview=(e.text or "")[:220] if getattr(e, "text", None) else None,
                section_hint=getattr(e, "section_hint", None),
            ).model_dump()
        )

    # Confidence heuristic: average of top few evidence scores (bounded 0..1).
    scores = sorted([float(e.score) for e in rep.evidence_items if e.score is not None], reverse=True)[:6]
    confidence = float(sum(scores) / len(scores)) if scores else 0.0

    doc_ids = {str(p.get("doc_id", "")) for p in prov_rows if p.get("doc_id")}
    filenames = await fetch_doc_filenames(db, tenant_id, doc_ids)
    # Attach filename and keep existing snippet previews.
    for p in prov_rows:
        did = str(p.get("doc_id") or "")
        if did and did in filenames:
            p["filename"] = filenames[did]

    cache_payload: dict[str, Any] = {
        "text": text,
        "confidence": confidence,
        "provenance": prov_rows,
        "style_profile": json.loads(style_profile.model_dump_json()),
        "ai_level": int(ai_norm.get("ai_level", ai_level)),
        "ai_percent": int(ai_norm.get("ai_percent", ai_level_to_percent(int(ai_level)))),
        "pipeline": "agentic",
        "fallback_used": False,
        "interference_level": interference_level,
    }
    if inspector_meta:
        cache_payload["inspector"] = inspector_meta
    if agentic_clamp_report.has_clamps:
        # Surface the clamp on the inspector payload so the API consumer can
        # render a "N notes dropped from this section" badge without needing
        # to thread a new metrics field through every persistence layer.
        cache_payload["bullet_clamps"] = agentic_clamp_report.to_dict()
    section_cache.set(cache_key, cache_payload)
    await _persist_section(
        db=db,
        report=report,
        section_code=template_id,
        text=text,
        confidence=confidence,
        provenance=prov_rows,
        cached=False,
        mode=GenerationMode.generate,
        style_profile=style_profile,
        ai_level=int(ai_norm.get("ai_level", ai_level)),
        ai_percent=int(ai_norm.get("ai_percent")) if ai_norm.get("ai_percent") is not None else None,
        pipeline="agentic",
        fallback_used=False,
        inspector=inspector_meta,
        interference_level=interference_level,
        mark_report_complete=mark_report_complete,
        bullet_clamps=(
            agentic_clamp_report.to_dict() if agentic_clamp_report.has_clamps else None
        ),
    )


# ── PROOFREAD mode ─────────────────────────────────────────────────────────────

async def _run_proofread(
    db: AsyncSession,
    report: Report,
    tenant_id: str,
    template_id: str,
    bullets: list[str],
    ai_level: int = AILevel.balanced,
    ai_percent: int | None = None,
    retrieval_level: str = "paragraph",
    reference_document_ids: list[str] | None = None,
    interference_level: str | None = None,
    mark_report_complete: bool = True,
) -> None:
    """Proofread an existing generated section for grammar and style."""
    logger.info(
        "Proofread start report=%s section=%s tenant=%s",
        report.id,
        template_id,
        tenant_id,
    )
    existing_text = await _get_existing_section_text(db, report.id, template_id)
    # Build style profile once and reuse it for both the optional seed generation
    # and the actual proofread call — avoids two file-cache reads per request.
    style_profile = await _get_or_build_style_profile(tenant_id)
    if not existing_text:
        logger.info("No existing text for proofread — generating seed text for section=%s", template_id)
        existing_text, _, _, _, _, _ = await _generate_section_text(
            tenant_id=tenant_id,
            template_id=template_id,
            bullets=bullets,
            style_profile=style_profile,
            ai_level=ai_level,
            ai_percent=ai_percent,
            primary_document_id=report.document_id,
            reference_document_ids=reference_document_ids,
            draft_paragraph=None,
            report_id=str(report.id),
            retrieval_level=retrieval_level,
            db=db,
            report_survey_level=report.survey_level,
            interference_level=interference_level,
        )
    proofread_temp, proofread_hint = _mode_controls(ai_level, GenerationMode.proofread, ai_percent=ai_percent)

    # Strip any editor-notes block appended by a previous proofread pass so the
    # LLM receives only the clean body text, not accumulated annotation cruft.
    clean_text = existing_text.split("\n\n[Editor notes:")[0].strip()

    proofread_output = await gen_llm.proofread(
        text=clean_text,
        bullets=bullets,
        style_profile=style_profile,
        temperature=proofread_temp,
        creativity_hint=proofread_hint,
    )

    # Split corrected text from editor notes
    if "---NOTES---" in proofread_output:
        corrected_text, notes = proofread_output.split("---NOTES---", 1)
        corrected_text = corrected_text.strip()
        notes_block = f"\n\n[Editor notes: {notes.strip()}]"
    else:
        corrected_text = proofread_output.strip()
        notes_block = ""

    # Non-invention enforcement on the proofread output. Proofread is mostly
    # cosmetic (grammar / style) but the LLM can still rewrite "rear elevation"
    # as "10 Kingsley Avenue" if it decides to be helpful — without this guard
    # the proofread mode bypasses the same protection that generate has, and
    # can convert a verified section into a hallucinated one.
    corrected_text = await async_enforce_verify(
        text=corrected_text,
        bullets=bullets,
        snippets=[],
        openai_api_key=settings.openai_api_key or "",
        model=settings.chat_model,
    )

    final_text = corrected_text + notes_block

    await _persist_section(
        db=db,
        report=report,
        section_code=template_id,
        text=final_text,
        confidence=0.95,
        provenance=[],
        cached=False,
        mode=GenerationMode.proofread,
        style_profile=style_profile,
        ai_level=int(_ai_level_to_params(ai_level, ai_percent=ai_percent).get("ai_level", ai_level)),
        ai_percent=ai_percent,
        interference_level=interference_level,
        mark_report_complete=mark_report_complete,
    )
    logger.info("Proofread complete for section=%s report=%s", template_id, report.id)


# ── ENHANCE mode ───────────────────────────────────────────────────────────────

async def _run_enhance(
    db: AsyncSession,
    report: Report,
    tenant_id: str,
    template_id: str,
    bullets: list[str],
    force_regenerate: bool,
    ai_level: int = AILevel.balanced,
    ai_percent: int | None = None,
    retrieval_level: str = "paragraph",
    reference_document_ids: list[str] | None = None,
    interference_level: str | None = None,
    mark_report_complete: bool = True,
) -> None:
    """Expand an existing section with broader technical evidence."""
    logger.info(
        "Enhance start report=%s section=%s tenant=%s",
        report.id,
        template_id,
        tenant_id,
    )
    existing_text = await _get_existing_section_text(db, report.id, template_id)
    # Build style profile once and reuse — avoids two cache reads per enhance call.
    style_profile = await _get_or_build_style_profile(tenant_id)
    if not existing_text:
        logger.info("No existing text for enhance — generating seed text for section=%s", template_id)
        existing_text, _, _, _, _, _ = await _generate_section_text(
            tenant_id=tenant_id,
            template_id=template_id,
            bullets=bullets,
            style_profile=style_profile,
            ai_level=ai_level,
            ai_percent=ai_percent,
            primary_document_id=report.document_id,
            reference_document_ids=reference_document_ids,
            draft_paragraph=None,
            report_id=str(report.id),
            retrieval_level=retrieval_level,
            db=db,
            report_survey_level=report.survey_level,
            interference_level=interference_level,
        )

    # Use a broader query for enhancement — more technical evidence
    query = " ".join(bullets) + " technical details construction condition"
    candidates = await retrieve_for_report_unified(
        query=query,
        tenant_id=tenant_id,
        primary_document_id=report.document_id,
        secondary_document_ids=reference_document_ids,
        k=settings.retrieval_top_k,
    )
    candidates = await filter_search_results_by_survey_level(
        db,
        tenant_id=tenant_id,
        results=candidates,
        report_survey_level=report.survey_level,
        primary_document_id=report.document_id,
        reference_document_ids=list(reference_document_ids or []),
    )
    top_results = rerank(query=query, results=candidates, top_n=min(5, settings.rerank_top_n + 2))
    snippets = [r.text for r in top_results]
    enhance_temp, enhance_hint = _mode_controls(ai_level, GenerationMode.enhance, ai_percent=ai_percent)

    # Strip any editor-notes block from a prior proofread pass so the LLM
    # receives only clean body text — mirrors the same guard in _run_proofread.
    clean_text = existing_text.split("\n\n[Editor notes:")[0].strip()
    enhanced_text = await gen_llm.enhance(
        text=clean_text,
        bullets=bullets,
        snippets=snippets,
        style_profile=style_profile,
        temperature=enhance_temp,
        creativity_hint=enhance_hint,
    )

    final_text = await async_enforce_verify(
        text=enhanced_text,
        bullets=bullets,
        snippets=snippets,
        openai_api_key=settings.openai_api_key,
        model=settings.chat_model,
    )

    provenance = [
        Provenance(doc_id=r.doc_id, chunk_id=r.chunk_id, score=round(r.score, 4)).model_dump()
        for r in top_results
    ]
    confidence = (
        sum(r.rerank_score for r in top_results) / len(top_results) if top_results else 0.0
    )
    doc_ids = {str(p.get("doc_id", "")) for p in provenance if p.get("doc_id")}
    filenames = await fetch_doc_filenames(db, tenant_id, doc_ids)
    provenance = attach_snippet_metadata(provenance, list(top_results), filenames)

    await _persist_section(
        db=db,
        report=report,
        section_code=template_id,
        text=final_text,
        confidence=confidence,
        provenance=provenance,
        cached=False,
        mode=GenerationMode.enhance,
        style_profile=style_profile,
        ai_level=int(_ai_level_to_params(ai_level, ai_percent=ai_percent).get("ai_level", ai_level)),
        ai_percent=ai_percent,
        interference_level=interference_level,
        mark_report_complete=mark_report_complete,
    )
    logger.info("Enhanced section=%s for report=%s", template_id, report.id)


# ── Persist helper ─────────────────────────────────────────────────────────────

async def _persist_section(
    db: AsyncSession,
    report: Report,
    section_code: str,
    text: str,
    confidence: float,
    provenance: list[dict[str, Any]],
    cached: bool,
    mode: str = GenerationMode.generate,
    style_profile: WritingStyleProfile | None = None,
    ai_level: int = AILevel.balanced,
    ai_percent: int | None = None,
    measured_ai_percent: int | None = None,
    verbatim_overlap: float | None = None,
    pipeline: str | None = None,
    fallback_used: bool | None = None,
    inspector: dict[str, Any] | None = None,
    interference_level: str | None = None,
    mark_report_complete: bool = True,
    bullet_clamps: dict[str, Any] | None = None,
    coverage_metrics: dict[str, Any] | None = None,
    citation_audit: dict[str, Any] | None = None,
) -> None:
    """Upsert a section row and optionally mark the report as complete.

    Args:
        db: Active async database session.
        report: Parent ``Report`` ORM object.
        section_code: RICS section code.
        text: Generated / proofread / enhanced section text.
        confidence: Average rerank score.
        provenance: List of provenance dicts.
        cached: Whether the result came from cache.
        mode: Generation mode used.
        style_profile: Style profile applied (if any).
        inspector: Optional OpenAI inspector loop artifacts (stored under ``meta.inspector``).
    """
    from app.db.database import is_sqlite_database

    existing = await db.execute(
        select(ReportSection).where(
            ReportSection.report_id == report.id,
            ReportSection.section_code == section_code,
        )
    )
    section_orm = existing.scalars().first()

    meta: dict[str, Any] = {}
    lvl, pct, _rag_only = _normalise_ai_controls(ai_level, ai_percent)
    transparency = compute_ai_transparency(
        mode,
        int(lvl),
        ai_percent=int(pct),
        measured_ai_percent=int(measured_ai_percent) if measured_ai_percent is not None else None,
    )
    meta = {
        "mode": mode,
        "ai_level": int(lvl),
        "ai_percent": int(pct),
        "ai_transparency": transparency,
    }
    if measured_ai_percent is not None:
        meta["measured_ai_percent"] = int(measured_ai_percent)
    if verbatim_overlap is not None:
        meta["verbatim_overlap"] = float(verbatim_overlap)
    if pipeline is not None:
        meta["pipeline"] = str(pipeline)
    if fallback_used is not None:
        meta["fallback_used"] = bool(fallback_used)
    il_norm = _normalise_interference_level(interference_level)
    if il_norm is not None:
        meta["interference_level"] = il_norm
    meta["word_count"] = len((text or "").split())
    meta["generated_at"] = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
    if style_profile is not None:
        meta["style_profile"] = json.loads(style_profile.model_dump_json())
    if inspector:
        meta["inspector"] = inspector
    if bullet_clamps:
        meta["bullet_clamps"] = bullet_clamps
    if coverage_metrics:
        meta["notes_coverage"] = coverage_metrics
    if citation_audit:
        meta["citation_audit"] = citation_audit
    provenance_json = json.dumps({"sources": provenance, "meta": meta})

    if section_orm is None and not is_sqlite_database():
        from sqlalchemy.dialects.postgresql import insert as pg_insert

        import uuid as _uuid

        stmt = (
            pg_insert(ReportSection)
            .values(
                id=str(_uuid.uuid4()),
                report_id=report.id,
                section_code=section_code,
                text=text,
                confidence=confidence,
                provenance=provenance_json,
                cached=cached,
            )
            .on_conflict_do_update(
                index_elements=["report_id", "section_code"],
                set_={
                    "text": text,
                    "confidence": confidence,
                    "provenance": provenance_json,
                    "cached": cached,
                },
            )
        )
        await db.execute(stmt)
        if mark_report_complete:
            report.status = ReportStatus.complete
            report.error_message = None
            report.generation_started_at = None
            report.generation_section_total = None
        await db.commit()
        return

    if section_orm is None:
        section_orm = ReportSection(
            report_id=report.id,
            section_code=section_code,
        )
        db.add(section_orm)

    section_orm.text = text
    section_orm.confidence = confidence
    section_orm.cached = cached
    logger.debug(
        "_persist_section: ai_level=%r ai_percent=%r -> lvl=%r pct=%r section=%s",
        ai_level,
        ai_percent,
        lvl,
        pct,
        section_code,
    )
    logger.debug(
        "transparency: ai_involvement_percent=%r",
        transparency.get("ai_involvement_percent"),
    )
    section_orm.provenance = provenance_json

    if mark_report_complete:
        report.status = ReportStatus.complete
        report.error_message = None
        report.generation_started_at = None
        report.generation_section_total = None
    await db.commit()


async def persist_agentic_full_report(
    db: AsyncSession,
    report: Report,
    *,
    tenant_id: str,
    sections_payload: dict[str, Any],
    style_profile: WritingStyleProfile,
    ai_percent: int,
    ai_level: int = AILevel.balanced,
    interference_level: str | None = None,
) -> None:
    """Persist ``generate_full_report`` output into ``report_sections`` and finalize status."""
    failed: list[str] = []
    codes = list(sections_payload.keys())
    for code in codes:
        payload = sections_payload.get(code) or {}
        text = str(payload.get("report_text") or "").strip()
        if not text:
            failed.append(code)
            continue
        evidence = payload.get("evidence_items") or []
        provenance: list[dict[str, Any]] = []
        scores: list[float] = []
        for ev in evidence:
            if not isinstance(ev, dict):
                continue
            row = {
                k: ev[k]
                for k in ("text", "doc_id", "chunk_id", "source", "section_hint", "kb")
                if ev.get(k) is not None
            }
            if row:
                provenance.append(row)
            if ev.get("score") is not None:
                try:
                    scores.append(float(ev["score"]))
                except (TypeError, ValueError):
                    pass
        confidence = sum(scores) / len(scores) if scores else 0.0
        inspector_meta = payload.get("inspector")
        inspector = inspector_meta if isinstance(inspector_meta, dict) else None
        await _persist_section(
            db=db,
            report=report,
            section_code=code,
            text=text,
            confidence=confidence,
            provenance=provenance,
            cached=False,
            mode=GenerationMode.generate,
            style_profile=style_profile,
            ai_level=ai_level,
            ai_percent=ai_percent,
            pipeline="agentic",
            fallback_used=False,
            inspector=inspector,
            interference_level=interference_level,
            mark_report_complete=False,
        )
    if not codes:
        report.status = ReportStatus.failed
        report.generation_started_at = None
        report.error_message = "Agentic generation produced no sections."
    else:
        _finalize_multi_section_report(
            report,
            failure_count=len(failed),
            total=len(codes),
            phase="agentic",
        )
    await db.commit()
    logger.info(
        "Persisted agentic full report id=%s sections=%d failed=%d",
        report.id,
        len(codes),
        len(failed),
    )


async def run_agentic_full_report_job(
    report_id: str,
    tenant_id: str,
    *,
    bullets_by_section: dict[str, list[str]],
    ai_percent: int,
    retrieval_level: str,
    reference_document_ids: list[str] | None,
    similarity_scan: bool,
    peer_sections: dict[str, str],
    similarity_exclude_document_ids: list[str] | None,
    interference_level: str | None,
) -> None:
    """Background job for POST /agentic/generate (owns its DB session)."""
    from app.agentic.agents import generate_full_report

    try:
        if not await abort_generation_if_report_invalid(
            report_id,
            tenant_id,
            reason="Report not found or access denied (agentic job).",
        ):
            return

        factory = get_session_factory()
        async with factory() as db:
            report = await db.get(Report, report_id)
            if report is None or report.tenant_id != tenant_id:
                await mark_report_generation_failed(
                    report_id,
                    tenant_id,
                    "Report not found or access denied (agentic job).",
                )
                return
            try:
                style_profile = await _get_or_build_style_profile(tenant_id)
                result = await generate_full_report(
                    db=db,
                    tenant_id=tenant_id,
                    primary_document_id=report.document_id,
                    bullets_by_section=bullets_by_section,
                    style_profile=style_profile,
                    ai_percent=ai_percent,
                    retrieval_level=retrieval_level,
                    reference_document_ids=reference_document_ids,
                    similarity_scan=similarity_scan,
                    peer_sections=peer_sections,
                    similarity_exclude_document_ids=similarity_exclude_document_ids,
                    survey_level=report.survey_level,
                )
                await persist_agentic_full_report(
                    db,
                    report,
                    tenant_id=tenant_id,
                    sections_payload=result.get("sections") or {},
                    style_profile=style_profile,
                    ai_percent=ai_percent,
                    interference_level=interference_level,
                )
            except Exception as exc:  # noqa: BLE001
                logger.exception("Agentic background job failed report=%s", report_id)
                await mark_report_generation_failed(report_id, tenant_id, str(exc))
    except Exception as exc:  # noqa: BLE001
        logger.exception("Unhandled agentic job error report=%s", report_id)
        await mark_report_generation_failed(report_id, tenant_id, str(exc))
    finally:
        await finalize_generation_status_if_stuck(report_id, tenant_id)