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, verbatim_overlap_ratio
from app.generator.prompts import verbatim_ratio_target
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.personalised_rag import (
    kb_style_fallback_allowed,
    resolve_retrieval_doc_allowlist,
    tenant_has_personal_library,
)
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_vision import enrich_bullets_with_section_photos
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 = 100) -> list[str]:
    """Normalise user bullets and clamp to an operational max.

    We accept long, messy note dumps (including routed notes) but generation
    prompts and downstream models have practical limits. This function:
    - trims whitespace and leading bullet markers
    - drops empty lines
    - de-duplicates while preserving order
    - clamps to max_items
    """
    if not bullets:
        return []
    out: list[str] = []
    seen: set[str] = set()
    for raw in bullets:
        t = str(raw or "").strip()
        if not t:
            continue
        t = re.sub(r"^[-•*]+\s*", "", t).strip()
        if not t:
            continue
        key = t.lower()
        if key in seen:
            continue
        seen.add(key)
        out.append(t)
        if len(out) >= max_items:
            break
    return out


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() + "…"
    return (
        f"{heading}\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 _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
    if "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"):
        parts.append(
            f"- Property type: {identity['property_type']} (do not describe as flats/communal unless explicitly stated)"
        )
    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."
    )
    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 three-tier fallback.

    Priority order:
    1. User's cached profile (built from their own uploaded documents).
    2. Fresh analysis of the user's currently indexed documents (caches result).
    3. Reference profile built from the KB corpus — only when personalised RAG is
       off or the tenant has not uploaded any completed reports yet.

    The cache is invalidated on every new upload, so style always reflects the
    user's current document 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 — analyse the user's own uploaded documents
    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

    has_library = await tenant_has_personal_library(tenant_id)
    if kb_style_fallback_allowed(tenant_id, has_personal_library=has_library):
        kb_tenant = settings.knowledge_base_tenant_id
        kb_profile = style_cache.get(kb_tenant)
        if kb_profile:
            logger.info(
                "No indexed uploads for tenant=%s — using reference style profile from KB",
                tenant_id,
            )
            return kb_profile

    logger.info(
        "No style samples for tenant=%s (personal_library=%s) — using mock style profile",
        tenant_id,
        has_library,
    )
    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,
                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.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,
                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 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 multi_section_parallel_enabled

            if multi_section_parallel_enabled():
                await _run_multi_section_parallel(**multi_kwargs)
            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.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
    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,
) -> None:
    """Run multiple sections concurrently (async pipeline)."""
    import structlog

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

    async def _one(section_code: str) -> None:
        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
    for section_code in section_codes:
        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,
                )

    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 _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,
        )

    bullets = _clean_and_clamp_bullets(list(bullets or []), max_items=2000)
    bullets_for_generation = _clean_and_clamp_bullets(bullets, max_items=100)

    if not bullets_for_generation:
        # No notes for this section: persist a blank section so the UI can
        # prompt the user to fill it in immediately (inline editor).
        return "", [], 0.0, [], [], {"requested_ai_percent": int(level_params.get("ai_percent", 50))}

    async def _expand_bullets_coro() -> list[str]:
        """Notes expansion; OpenAI path runs in a thread so the event loop stays responsive."""
        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_for_generation,
                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_for_generation,
            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

    expanded_bullets = await _expand_bullets_coro()
    if report_id:
        expanded_bullets = await enrich_bullets_with_section_photos(
            db,
            tenant_id=tenant_id,
            report_id=str(report_id),
            section_code=template_id,
            bullets=expanded_bullets,
            survey_level=report_survey_level,
        )

    # Use expanded bullets for internal query strings (and retrieval when enabled).
    query = " ".join(expanded_bullets)
    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] = []

    # Retrieval scope: personalised RAG uses the tenant's full ingested library;
    # strict_uploaded_only limits to docs attached to this report only.
    allowed = await resolve_retrieval_doc_allowlist(
        db,
        tenant_id,
        primary_document_id=primary_document_id,
        reference_document_ids=ref_ids,
        runtime_doc_ids=runtime_extra,
        strict_uploaded_only=strict_uploaded_only,
    )

    if not allowed:
        logger.info(
            "No retrieval corpus for tenant=%s section=%s — generation uses bullets/style only",
            tenant_id,
            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()
    section_std_prepared = ""
    if section_std_raw:
        section_std_prepared = (
            _redact_reference_snippet(section_std_raw).strip()
            if notes_only
            else 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 and not settings.personalised_style_rag_enabled:
        # Non-personalised runs may prepend packaged standard paragraphs at low AI %.
        if section_std_prepared and ap_inv <= 12:
            para_snippets = [section_std_prepared] + [
                p for p in para_snippets if p.strip() != section_std_prepared
            ]

    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

    identity = _extract_property_identity(list(bullets_for_generation))
    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.
                return "", [], 0.0, [], [], {"requested_ai_percent": int(ap_branch)}

    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,
        )
    # Layer 1+2 grounding: regex fast pass + LLM context-aware grounding
    text = await async_enforce_verify(
        text=raw_text,
        bullets=bullets_for_generation,
        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=bullets_for_generation,
                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,
    }

    # 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_for_generation,
            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_for_generation,
            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))

    # 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_for_generation,
                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


# ── 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,
            )
            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)

    # 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,
    })
    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,
    )
    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", "agentic") or "").strip().lower()
    return raw if raw in ("agentic", "standard") else "agentic"


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", True)
    )
    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."""
    # 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,
        report_id=str(report.id),
    )
    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
    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,
    )


# ── 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,
) -> 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
    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
        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
    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,
                    report_id=str(report.id),
                )
                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)