backend/core/section_mapper.py

"""Generation orchestrator.

Every schema section is generated from the most comprehensive REFERENCE-tier block
retrieved from uploaded past reports. Surveyor notes trigger in-place fact updates
on that baseline only — no scratch prose generation. All interference levels share
the same text-anchored adaptation path.
"""

from __future__ import annotations

import asyncio
import logging
import re
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from typing import TypeAlias

from backend.config import settings
from backend.core import pii_scrubber, photo_store, template_discoverer
from backend.core.notes_parser import UNASSIGNED, parse_notes_to_sections
from backend.core.paragraph_retriever import (
    InterferenceLevel,
    RetrievalLevel,
    _uses_reference_tier,
    assemble_reference_baseline,
    fetch_complete_section_baseline,
    find_paragraph_by_topic,
    guess_report_section_from_topic,
    report_section_for_paragraph_id,
    retrieve_paragraphs_for_mapping,
)
from backend.core.interference import resolve_interference_level
from backend.core.reference_filter import build_reference_allowlist
from backend.core.rag_store import TIER_MASTER, TIER_REFERENCE, SearchHit
from backend.core.atomic_observations import split_atomic_observations
from backend.core.rics_canonical_l3 import (
    mapping_units_for_parent,
    ordered_parent_sections,
    valid_leaf_section_ids,
)
from backend.core.observation_matcher import partition_observations_for_baseline
from backend.core.paragraph_merge import bound_baseline_to_notes
from backend.core.composition_output import sanitize_section_prose
from backend.core.specified_resolver import resolve_specified_tokens
from backend.core.survey_notes import build_property_context, parse_notes
from backend.core.reference_mapper import map_reference_paragraph
from backend.core.source_attribution import format_reference_attribution
from backend.core.validation_orchestrator import (
    run_validation_batch,
    stabilize_section,
)
from backend.core.vision_analyzer import vision_observations_for_section
from backend.llm import openai_client
from backend.models.report import GeneratedSection, ReportResult
from backend.models.validation_loop import SectionValidationInput, SectionValidationResult
from backend.models.schema import TemplateSchema
from backend.models.section import SectionNote
from backend.prompts.notes_expander import build_expander_messages
from backend.utils.tenant_store import path_safe_section_id

logger = logging.getLogger(__name__)

SectionCompleteCallback: TypeAlias = Callable[[GeneratedSection], Awaitable[None] | None]
DEFAULT_SECTION_CONCURRENCY = 54

# Debug / intermediate dumps must use section IDs only (F2, M, D1) — never labels
# such as "Gas/Oil" or "service and terms of engagement", which break Windows paths.


def section_debug_filename(section_id: str, ext: str = "txt") -> str:
    """Filesystem-safe debug basename — canonical section ID, never the human label."""
    sid = path_safe_section_id(section_id)
    suffix = ext.lstrip(".") or "txt"
    return f"section_{sid}.{suffix}"

_UNMATCHED_TAG_RE = re.compile(
    r"\[UNMATCHED_OBSERVATION:\s*(.*?)\]",
    re.IGNORECASE | re.DOTALL,
)
_UNMATCHED_LINE_RE = re.compile(
    r"^\s*UNMATCHED_OBSERVATION:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)

_NO_RAG_PLACEHOLDER = (
    "[No past-report paragraph found for this section. Manual entry required.]"
)

_UNMATCHED_SECTION_HEADING = "### UNMATCHED_OBSERVATION"

_PHOTO_LIMITATIONS_PREFIX = "Photo limitations:"


def _photo_limitation_observations(photo_note: str | None) -> list[str]:
    """Extract vision limitation sentences from the photo analysis user note."""
    if not photo_note or _PHOTO_LIMITATIONS_PREFIX not in photo_note:
        return []
    raw = photo_note.split(_PHOTO_LIMITATIONS_PREFIX, 1)[1].strip()
    if not raw:
        return []
    parts = [part.strip() for part in raw.split(". ") if part.strip()]
    if not parts:
        return [raw]
    return [part if part.endswith(".") else f"{part}." for part in parts]


def _baseline_passthrough_result(
    section_id: str,
    baseline_paragraph: str,
) -> SectionValidationResult:
    """Skip semantic audit when no new observations were mapped."""
    return SectionValidationResult(
        section_id=section_id,
        status="STABILIZED",
        text=baseline_paragraph.strip(),
        iterations=0,
    )


def _section_has_selected_photos(
    tenant_id: str,
    draft_id: str | None,
    section_id: str,
) -> bool:
    if not draft_id:
        return False
    rows = photo_store.list_section_photos(tenant_id, draft_id, section_id)
    return any(r.selected_for_ai for r in rows)


def collect_active_section_ids(
    schema: TemplateSchema,
    by_id: dict[str, SectionNote],
    *,
    tenant_id: str,
    report_draft_id: str | None,
    only_section_ids: list[str] | None,
) -> list[str]:
    """Ordered canonical leaf IDs with surveyor notes and/or AI-selected photos."""
    active: set[str] = set()
    for sid, note in by_id.items():
        if sid.upper() == UNASSIGNED:
            continue
        if note.raw_observations or (note.text or "").strip():
            active.add(sid.upper())

    if report_draft_id:
        for parent in ordered_parent_sections(schema):
            for sec in mapping_units_for_parent(schema, parent.id):
                if _section_has_selected_photos(tenant_id, report_draft_id, sec.id):
                    active.add(sec.id.upper())

    allowed: set[str] | None = None
    if only_section_ids is not None:
        allowed = {
            (x or "").strip().upper()
            for x in only_section_ids
            if (x or "").strip()
        }

    ordered: list[str] = []
    for parent in ordered_parent_sections(schema):
        for sec in mapping_units_for_parent(schema, parent.id):
            sid = sec.id.upper()
            if sid not in active:
                continue
            if allowed is not None and sid not in allowed:
                continue
            ordered.append(sec.id)
    return ordered


def estimate_active_sections_from_generate_body(
    body: object,
    *,
    tenant_id: str,
    draft_id: str | None,
) -> list[str]:
    """Active subsection IDs for progress tracking before generation starts."""
    by_sec = getattr(body, "bullets_by_section", None) or {}
    template_id = (getattr(body, "template_id", None) or "").strip()
    bullets = list(getattr(body, "bullets", None) or [])
    template_ids = list(getattr(body, "template_ids", None) or [])

    def _has_bullets(section_code: str) -> bool:
        items = by_sec.get(section_code) or by_sec.get(section_code.upper()) or []
        if items and any((i or "").strip() for i in items):
            return True
        return section_code == template_id and any((b or "").strip() for b in bullets)

    codes: set[str] = set()
    if by_sec:
        for code, items in by_sec.items():
            c = (code or "").strip().upper()
            if c and any((i or "").strip() for i in items):
                codes.add(c)
    elif bullets and template_id:
        codes.add(template_id.upper())

    for code in template_ids:
        c = (code or "").strip()
        if c and _has_bullets(c):
            codes.add(c.upper())

    scan = set(codes)
    for code in template_ids:
        c = (code or "").strip().upper()
        if c:
            scan.add(c)
    if template_id:
        scan.add(template_id.upper())

    if draft_id:
        for sid in scan:
            if _section_has_selected_photos(tenant_id, draft_id, sid):
                codes.add(sid.upper())

    if template_ids:
        ordered: list[str] = []
        seen: set[str] = set()
        for code in template_ids:
            c = (code or "").strip().upper()
            if c in codes and c not in seen:
                seen.add(c)
                ordered.append(c)
        ordered.extend(sorted(codes - seen))
        return ordered
    return sorted(codes)


@dataclass
class _MapOutcome:
    text: str
    hits: list[SearchHit]
    no_rag_match: bool
    unmatched_observations: list[str] | None = None
    baseline_paragraph: str = ""


@dataclass
class _PendingValidatedSection:
    """Mapped draft awaiting judge-editor validation before payload commit."""

    section: GeneratedSection
    baseline_paragraph: str
    draft_text: str
    observations: list[str]
    template_paragraphs: list[str] = field(default_factory=list)


@dataclass
class _SectionMappingResult:
    """Outcome of the synchronous map phase for one section."""

    section: GeneratedSection | None = None
    pending: _PendingValidatedSection | None = None
    unmatched: list[str] = field(default_factory=list)


def _expand_notes(
    schema: TemplateSchema,
    notes_text: str,
    *,
    interference_level: InterferenceLevel,
) -> str:
    if not settings.notes_expansion_enabled or not openai_client.is_available():
        return notes_text
    try:
        return openai_client.chat_text(
            build_expander_messages(
                schema,
                notes_text,
                interference_level=interference_level,
            ),
            model=settings.mapping_model,
            max_tokens=settings.max_tokens_mapping,
        ) or notes_text
    except Exception as exc:  # noqa: BLE001
        logger.warning("Notes expansion failed (%s); using raw notes.", exc)
        return notes_text


def _observations_after_expand(
    raw_observations: list[str],
    notes_text: str,
    expanded_text: str,
) -> list[str]:
    if expanded_text.strip() != notes_text.strip():
        lines = [
            line.strip().lstrip("•-*·▸▹◆►").strip()
            for line in expanded_text.split("\n")
            if line.strip()
        ]
        return lines or raw_observations
    return raw_observations


def _reroute_unassigned_via_rag(
    tenant_id: str,
    schema: TemplateSchema,
    unassigned: SectionNote | None,
    by_id: dict[str, SectionNote],
    *,
    interference_level: InterferenceLevel,
    allowed_doc_keys: frozenset[str] | None = None,
) -> list[str]:
    """Match orphan note lines to REFERENCE paragraphs by topic (secondary gate).

    Notes bucketed as ``UNASSIGNED`` by anchor similarity in :mod:`notes_parser`
    may be promoted here when the top REFERENCE hit scores >=
    ``settings.confidence_threshold`` (0.72).
    """
    if unassigned is None or not unassigned.raw_observations:
        return []

    still_orphan: list[str] = []
    for block in unassigned.raw_observations:
        hits = find_paragraph_by_topic(
            tenant_id,
            [block],
            interference_level=interference_level,
            allowed_doc_keys=allowed_doc_keys,
            paragraph_section_id="",
        )
        if not hits or hits[0].score < settings.confidence_threshold:
            still_orphan.append(block)
            continue
        report_sid = report_section_for_paragraph_id(schema, hits[0].section_id)
        if report_sid is None:
            report_sid = guess_report_section_from_topic(schema, [block], hits[0].text)
        if report_sid is None or report_sid.upper() not in valid_leaf_section_ids():
            still_orphan.append(block)
            continue
        if report_sid in by_id:
            by_id[report_sid].raw_observations.append(block)
            by_id[report_sid].text = "\n".join(by_id[report_sid].raw_observations).strip()
        else:
            by_id[report_sid] = SectionNote(
                section_id=report_sid,
                raw_observations=[block],
                text=block,
            )
    return still_orphan


def _map_section(
    schema: TemplateSchema,
    tenant_id: str,
    section_title: str,
    section_id: str,
    observations: list[str],
    rating_value: str | None,
    *,
    interference_level: InterferenceLevel,
    retrieval_level: RetrievalLevel = "paragraph",
    allowed_doc_keys: frozenset[str] | None = None,
    property_context: dict | None = None,
) -> _MapOutcome:
    # The section_alias_map redirects report codes to the MASTER standard-paragraph
    # codes (e.g. firm bundle uses E-codes where canonical uses D-codes). An uploaded
    # past report may be authored with EITHER numbering, so for the reference tier we
    # try the canonical report code first and fall back to the alias. Applying only
    # the alias misroutes correctly-tagged references (e.g. D2 Roof -> E2 Walls).
    alias_id = schema.paragraph_section_id(section_id)
    if _uses_reference_tier(interference_level):
        candidate_ids = [section_id]
        if alias_id and alias_id != section_id:
            candidate_ids.append(alias_id)
    else:
        candidate_ids = [alias_id]
    query_obs = observations or [section_title]

    tier = TIER_REFERENCE if _uses_reference_tier(interference_level) else TIER_MASTER
    paragraph_id = candidate_ids[0]
    hits: list[SearchHit] = []
    baseline_text, baseline_hits = "", []
    for cid in candidate_ids:
        cand_hits = retrieve_paragraphs_for_mapping(
            tenant_id,
            section_label=section_title,
            paragraph_section_id=cid,
            observations=query_obs,
            interference_level=interference_level,
            retrieval_level=retrieval_level,
            allowed_doc_keys=allowed_doc_keys,
            property_context=property_context,
        )
        if not cand_hits:
            continue
        cand_text, cand_hits_used = assemble_reference_baseline(
            cand_hits,
            paragraph_section_id=cid,
            tenant_id=tenant_id,
            tier=tier,
            allowed_doc_keys=allowed_doc_keys,
            property_context=property_context,
        )
        if cand_text.strip():
            paragraph_id = cid
            hits = cand_hits
            baseline_text, baseline_hits = cand_text, cand_hits_used
            break

    if not hits or not baseline_text.strip():
        # Section-complete fallback: similarity surfaced nothing, but the section may
        # still exist in the index (weak query match / alias drift). Pull it directly
        # by metadata before declaring no-RAG, so a real past-report section is mapped
        # instead of degrading to a notes-only paragraph.
        for cid in candidate_ids:
            fb_text, fb_hits = fetch_complete_section_baseline(
                tenant_id,
                paragraph_section_id=cid,
                tier=tier,
                allowed_doc_keys=allowed_doc_keys,
                property_context=property_context,
            )
            if fb_text.strip():
                paragraph_id = cid
                hits = fb_hits
                baseline_text, baseline_hits = fb_text, fb_hits
                break

    if not hits or not baseline_text.strip():
        return _MapOutcome(text=_NO_RAG_PLACEHOLDER, hits=[], no_rag_match=True)

    # Per-note RAG gate: below confidence_threshold → UNMATCHED (never sent to LLM).
    atomic_obs = split_atomic_observations(observations)
    mappable, below_threshold = partition_observations_for_baseline(
        tenant_id,
        atomic_obs,
        baseline_text,
        paragraph_section_id=paragraph_id,
        interference_level=interference_level,
        allowed_doc_keys=allowed_doc_keys,
        report_section_id=section_id,
    )

    if mappable:
        mapped = map_reference_paragraph(
            baseline_text,
            mappable,
            schema,
            interference_level,
            section_id=section_id,
            section_title=section_title,
            rating_value=rating_value,
        )
        text = mapped or baseline_text
        result_baseline = baseline_text
    else:
        # No surveyor note mapped onto this section's past-report baseline. The
        # baseline describes a DIFFERENT property, so reproducing it verbatim would
        # assert facts the surveyor never recorded. Reduce it to notes-supported +
        # generic content; if nothing safe survives, author from the notes instead
        # of shipping another property's section.
        bounded = bound_baseline_to_notes(baseline_text, observations)
        if not bounded.strip():
            authored = _author_from_findings(observations)
            return _MapOutcome(
                text=authored or _NO_RAG_PLACEHOLDER,
                hits=[],
                no_rag_match=not bool(authored),
                unmatched_observations=below_threshold,
                baseline_paragraph="",
            )
        text = bounded
        result_baseline = bounded

    return _MapOutcome(
        text=text,
        hits=baseline_hits or hits[:1],
        no_rag_match=False,
        unmatched_observations=below_threshold,
        baseline_paragraph=result_baseline,
    )


def _format_unmatched_section_block(unmatched: list[str]) -> str:
    """Append structured unmatched block for template-schema export."""
    items = [u.strip() for u in unmatched if u.strip()]
    if not items:
        return ""
    bullets = "\n".join(f"* {item}" for item in items)
    return f"\n\n{_UNMATCHED_SECTION_HEADING}\n{bullets}"


def _extract_unmatched(mapped: str) -> tuple[str, list[str]]:
    tag_matches = [m.strip() for m in _UNMATCHED_TAG_RE.findall(mapped)]
    line_matches = [m.strip() for m in _UNMATCHED_LINE_RE.findall(mapped)]
    unmatched = tag_matches + line_matches
    cleaned = _UNMATCHED_TAG_RE.sub("", mapped)
    cleaned = _UNMATCHED_LINE_RE.sub("", cleaned).strip()
    return cleaned, unmatched


def _run_coroutine_sync(coro):
    """Execute an async coroutine from sync callers (tests, threadpool workers)."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    # Nested inside a running loop (e.g. async test harness): isolate with a fresh loop.
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(coro)
    finally:
        loop.close()


def _apply_validation_outcome(
    item: _PendingValidatedSection,
    outcome: SectionValidationResult | None,
) -> None:
    """Commit stabilized prose (or rollback) onto a pending mapped section."""
    if outcome is None:
        logger.error(
            "validation_batch_missing_result section=%s — rolling back to baseline",
            item.section.section_id,
        )
        item.section.text = item.baseline_paragraph
        item.section.status = "GROUNDING_REVIEW"
        item.section.grounding_passed = False
        return

    item.section.text = outcome.text
    if outcome.status == "STABILIZED":
        item.section.status = "OK"
        item.section.grounding_passed = True
    else:
        item.section.status = "GROUNDING_REVIEW"
        item.section.grounding_passed = False
        if outcome.failure:
            logger.warning(
                "validation_circuit_breaker section=%s reason=%s iterations=%s",
                item.section.section_id,
                outcome.failure.reason,
                outcome.iterations,
            )


async def _invoke_section_complete(
    callback: SectionCompleteCallback | None,
    section: GeneratedSection,
) -> None:
    if callback is None:
        return
    try:
        maybe_awaitable = callback(section)
        if asyncio.iscoroutine(maybe_awaitable):
            await maybe_awaitable
    except Exception:  # noqa: BLE001 — callback must not abort sibling sections
        logger.exception(
            "on_section_complete failed section=%s",
            section.section_id,
        )


def _prepare_section_mapping_sync(
    sec_id: str,
    *,
    schema: TemplateSchema,
    tenant_id: str,
    by_id: dict[str, SectionNote],
    id_to_unit: dict[str, object],
    report_draft_id: str | None,
    interference_level: InterferenceLevel,
    retrieval_level: RetrievalLevel,
    allowed_doc_keys: frozenset[str] | None,
    property_context: dict | None = None,
) -> _SectionMappingResult:
    """Blocking map phase for one section (safe to run in ``asyncio.to_thread``)."""
    sec = id_to_unit[sec_id]
    note = by_id.get(sec.id)  # type: ignore[union-attr]
    observations: list[str] = []
    rating_value: str | None = None
    shorthand_expanded: str | None = None
    raw_observations: list[str] = []

    if note and note.text.strip():
        notes_text = _expand_notes(
            schema, note.text, interference_level=interference_level
        )
        if notes_text.strip() != note.text.strip():
            shorthand_expanded = notes_text
        observations = _observations_after_expand(
            note.raw_observations, note.text, notes_text
        )
        rating_value = note.rating_value
        raw_observations = list(note.raw_observations)

    photo_obs, photo_note = vision_observations_for_section(
        tenant_id, report_draft_id, sec.id, sec.title  # type: ignore[union-attr]
    )
    limitation_obs = _photo_limitation_observations(photo_note)
    combined_observations = [*observations, *photo_obs, *limitation_obs]

    outcome = _map_section(
        schema,
        tenant_id,
        sec.title,  # type: ignore[union-attr]
        sec.id,  # type: ignore[union-attr]
        combined_observations,
        rating_value,
        interference_level=interference_level,
        retrieval_level=retrieval_level,
        allowed_doc_keys=allowed_doc_keys,
        property_context=property_context,
    )

    if outcome.no_rag_match:
        # Notes-first fallback: when no past-report baseline was retrieved but the
        # surveyor recorded findings for this section, author a clean paragraph
        # from those findings instead of leaving a dead tombstone. This only runs
        # on the otherwise-empty path, so it cannot affect sections that already
        # map successfully.
        notes_authored = _author_from_findings(combined_observations)
        if notes_authored:
            notes_authored = resolve_specified_tokens(
                notes_authored,
                section_code=sec.id,  # type: ignore[union-attr]
                property_context=property_context,
            )
            return _SectionMappingResult(
                section=GeneratedSection(
                    section_id=sec.id,  # type: ignore[union-attr]
                    title=sec.title,  # type: ignore[union-attr]
                    text=notes_authored,
                    rating_value=rating_value if schema.rating_system.detected else None,
                    status="NOTES_ONLY",
                    notes=(
                        "Authored from surveyor notes — no past-report baseline was "
                        "retrieved and no grounding audit was performed. Requires "
                        "surveyor review before issue."
                    ),
                    # Not grounding-audited against a baseline: keep False so no
                    # downstream path can treat notes-authored prose as verified.
                    grounding_passed=False,
                    shorthand_expanded=shorthand_expanded,
                ),
            )
        return _SectionMappingResult(
            section=GeneratedSection(
                section_id=sec.id,  # type: ignore[union-attr]
                title=sec.title,  # type: ignore[union-attr]
                text=outcome.text,
                rating_value=rating_value if schema.rating_system.detected else None,
                status="NO_RAG_MATCH",
                notes=(
                    "; ".join(raw_observations)
                    if raw_observations
                    else "No past-report paragraph retrieved for this section."
                ),
                unmatched_observations=raw_observations,
                grounding_passed=False,
                shorthand_expanded=shorthand_expanded,
            ),
        )

    mapped, unmatched = _extract_unmatched(outcome.text)
    section_unmatched = list(outcome.unmatched_observations or []) + unmatched

    reference_paragraphs = [h.text for h in outcome.hits]
    ref_sources, rag_sources = format_reference_attribution(
        outcome.hits, schema, max_sources=3
    )
    mappable_observations = [
        o for o in combined_observations if o not in section_unmatched
    ]
    for lim in limitation_obs:
        if lim.strip() and lim not in mappable_observations:
            mappable_observations.append(lim)
    draft_text = sanitize_section_prose(mapped)
    draft_text = resolve_specified_tokens(
        draft_text,
        section_code=sec.id,  # type: ignore[union-attr]
        property_context=property_context,
    )
    baseline_paragraph = (outcome.baseline_paragraph or reference_paragraphs[0] or "").strip()

    section_notes_msg = ""
    if section_unmatched:
        section_notes_msg = "; ".join(section_unmatched)
    elif photo_note:
        section_notes_msg = photo_note
    elif not observations and not photo_obs:
        section_notes_msg = (
            "Past-report paragraph unchanged (no surveyor notes — complete manually)."
        )
    elif photo_obs and not observations:
        section_notes_msg = (
            f"Content mapped from {len(photo_obs)} photo observation(s)."
        )

    rating = rating_value if schema.rating_system.detected else None
    generated = GeneratedSection(
        section_id=sec.id,  # type: ignore[union-attr]
        title=sec.title,  # type: ignore[union-attr]
        text=draft_text,
        rating_value=rating,
        status="OK",
        notes=section_notes_msg,
        rag_sources=rag_sources,
        reference_sources=ref_sources,
        grounding_passed=True,
        unmatched_observations=section_unmatched,
        shorthand_expanded=shorthand_expanded,
    )
    return _SectionMappingResult(
        pending=_PendingValidatedSection(
            section=generated,
            baseline_paragraph=baseline_paragraph,
            draft_text=draft_text,
            observations=mappable_observations,
            template_paragraphs=reference_paragraphs,
        ),
        unmatched=section_unmatched,
    )


def _author_from_findings(observations: list[str]) -> str:
    """Reformat surveyor observation lines into clean prose without inventing content.

    Used only on the otherwise-tombstone path (no past-report baseline). Every
    word originates from the surveyor's own notes — nothing is added.
    """
    items = [o.strip().strip("-*•·").strip() for o in observations if o and o.strip()]
    sentences: list[str] = []
    for item in items:
        if not item:
            continue
        sentence = item[0].upper() + item[1:]
        if sentence[-1] not in ".!?":
            sentence += "."
        sentences.append(sentence)
    return " ".join(sentences).strip()


def _failed_section(
    sec_id: str,
    *,
    title: str,
    error: str,
) -> GeneratedSection:
    return GeneratedSection(
        section_id=sec_id,
        title=title,
        text=_NO_RAG_PLACEHOLDER,
        status="GROUNDING_REVIEW",
        notes=f"Section generation failed: {error}",
        grounding_passed=False,
    )


async def _process_one_section(
    sec_id: str,
    *,
    schema: TemplateSchema,
    tenant_id: str,
    by_id: dict[str, SectionNote],
    id_to_unit: dict[str, object],
    report_draft_id: str | None,
    interference_level: InterferenceLevel,
    retrieval_level: RetrievalLevel,
    allowed_doc_keys: frozenset[str] | None,
    property_context: dict | None = None,
) -> tuple[GeneratedSection, list[str]]:
    """Map and validate one section; errors are isolated to a fallback section."""
    sec = id_to_unit.get(sec_id)
    title = sec.title if sec is not None else sec_id  # type: ignore[union-attr]
    try:
        mapping = await asyncio.to_thread(
            _prepare_section_mapping_sync,
            sec_id,
            schema=schema,
            tenant_id=tenant_id,
            by_id=by_id,
            id_to_unit=id_to_unit,
            report_draft_id=report_draft_id,
            interference_level=interference_level,
            retrieval_level=retrieval_level,
            allowed_doc_keys=allowed_doc_keys,
            property_context=property_context,
        )
        if mapping.section is not None:
            return mapping.section, mapping.unmatched

        pending = mapping.pending
        if pending is None:
            return _failed_section(sec_id, title=title, error="missing_map_result"), []

        if not pending.observations:
            logger.info(
                "baseline_passthrough_no_observations section=%s — skipping semantic audit",
                pending.section.section_id,
            )
            outcome = _baseline_passthrough_result(
                pending.section.section_id,
                pending.baseline_paragraph,
            )
        else:
            validation_input = SectionValidationInput(
                section_id=pending.section.section_id,
                section_label=pending.section.title,
                baseline_paragraph=pending.baseline_paragraph,
                draft_text=pending.draft_text,
                observations=pending.observations,
                template_paragraphs=pending.template_paragraphs,
            )
            outcome = await stabilize_section(validation_input)
        _apply_validation_outcome(pending, outcome)
        return pending.section, mapping.unmatched
    except Exception as exc:  # noqa: BLE001
        logger.exception("section_processing_failed section=%s", sec_id)
        return _failed_section(sec_id, title=title, error=str(exc)), []


async def _apply_validation_batch(
    pending: list[_PendingValidatedSection],
) -> None:
    """Run the judge-editor loop over mapped drafts and commit stabilized prose."""
    if not pending:
        return

    to_validate: list[_PendingValidatedSection] = []
    stabilized_count = 0
    circuit_breaker_count = 0

    for item in pending:
        if not item.observations:
            logger.info(
                "baseline_passthrough_no_observations section=%s — skipping semantic audit",
                item.section.section_id,
            )
            _apply_validation_outcome(
                item,
                _baseline_passthrough_result(
                    item.section.section_id,
                    item.baseline_paragraph,
                ),
            )
            stabilized_count += 1
            continue
        to_validate.append(item)

    results: list[SectionValidationResult] = []
    if to_validate:
        inputs = [
            SectionValidationInput(
                section_id=item.section.section_id,
                section_label=item.section.title,
                baseline_paragraph=item.baseline_paragraph,
                draft_text=item.draft_text,
                observations=item.observations,
                template_paragraphs=item.template_paragraphs,
            )
            for item in to_validate
        ]
        results = await run_validation_batch(inputs, concurrency=54)

    by_id = {r.section_id.upper(): r for r in results}

    for item in to_validate:
        outcome = by_id.get(item.section.section_id.upper())
        _apply_validation_outcome(item, outcome)
        if outcome is not None and outcome.status == "STABILIZED":
            stabilized_count += 1
        else:
            circuit_breaker_count += 1

    total = len(pending)
    logger.info(
        "validation_batch_complete total_sections=%d stabilized=%d "
        "circuit_breaker_triggered=%d",
        total,
        stabilized_count,
        circuit_breaker_count,
    )


async def generate_report_async(
    tenant_id: str,
    raw_notes: str,
    *,
    property_type: str = "",
    tenure: str = "",
    interference_level: InterferenceLevel | None = None,
    survey_level: int = 3,
    retrieval_level: RetrievalLevel = "paragraph",
    report_draft_id: str | None = None,
    only_section_ids: list[str] | None = None,
    reference_document_ids: list[str] | None = None,
    strict_uploaded_only: bool = False,
    on_section_complete: SectionCompleteCallback | None = None,
    section_concurrency: int = DEFAULT_SECTION_CONCURRENCY,
) -> ReportResult:
    schema = template_discoverer.ensure_canonical_schema(tenant_id)

    interference_level = resolve_interference_level(interference_level, survey_level)

    allowed_doc_keys = build_reference_allowlist(
        tenant_id,
        reference_document_ids,
        strict_uploaded_only=strict_uploaded_only,
    )

    section_notes = parse_notes_to_sections(raw_notes, schema)
    by_id = {n.section_id: n for n in section_notes if n.section_id != UNASSIGNED}
    unassigned_note = next((n for n in section_notes if n.section_id == UNASSIGNED), None)

    # Structured note extraction → property context for the retrieval guard (Fix 1/3).
    survey_notes = parse_notes(raw_notes)
    property_context = build_property_context(
        survey_notes, property_type=property_type, tenure=tenure
    )

    orphan = _reroute_unassigned_via_rag(
        tenant_id,
        schema,
        unassigned_note,
        by_id,
        interference_level=interference_level,
        allowed_doc_keys=allowed_doc_keys,
    )

    result = ReportResult(
        tenant_id=tenant_id,
        schema_version=schema.version,
        property_type=property_type,
        tenure=tenure,
    )
    unmatched_global: list[str] = []

    active_section_ids = collect_active_section_ids(
        schema,
        by_id,
        tenant_id=tenant_id,
        report_draft_id=report_draft_id,
        only_section_ids=only_section_ids,
    )
    result.active_section_count = len(active_section_ids)

    id_to_unit = {
        sec.id: sec
        for parent in ordered_parent_sections(schema)
        for sec in mapping_units_for_parent(schema, parent.id)
    }

    if active_section_ids:
        limit = max(1, min(section_concurrency, len(active_section_ids)))
        semaphore = asyncio.Semaphore(limit)
        progress_lock = asyncio.Lock()

        async def _run_one(sec_id: str) -> tuple[str, GeneratedSection, list[str]]:
            async with semaphore:
                try:
                    section, unmatched = await _process_one_section(
                        sec_id,
                        schema=schema,
                        tenant_id=tenant_id,
                        by_id=by_id,
                        id_to_unit=id_to_unit,
                        report_draft_id=report_draft_id,
                        interference_level=interference_level,
                        retrieval_level=retrieval_level,
                        allowed_doc_keys=allowed_doc_keys,
                        property_context=property_context,
                    )
                    await _invoke_section_complete(on_section_complete, section)
                    async with progress_lock:
                        result.processed_section_count += 1
                    return sec_id, section, unmatched
                except Exception as exc:  # noqa: BLE001
                    logger.exception("unhandled_section_task_failure section=%s", sec_id)
                    sec = id_to_unit.get(sec_id)
                    fallback = _failed_section(
                        sec_id,
                        title=sec.title if sec is not None else sec_id,  # type: ignore[union-attr]
                        error=str(exc),
                    )
                    await _invoke_section_complete(on_section_complete, fallback)
                    async with progress_lock:
                        result.processed_section_count += 1
                    return sec_id, fallback, []

        gathered = await asyncio.gather(
            *[_run_one(sec_id) for sec_id in active_section_ids],
            return_exceptions=True,
        )

        for index, item in enumerate(gathered):
            sec_id = active_section_ids[index]
            if isinstance(item, Exception):
                logger.exception(
                    "section_gather_failure section=%s",
                    sec_id,
                    exc_info=item,
                )
                sec = id_to_unit.get(sec_id)
                fallback = _failed_section(
                    sec_id,
                    title=sec.title if sec is not None else sec_id,  # type: ignore[union-attr]
                    error=str(item),
                )
                await _invoke_section_complete(on_section_complete, fallback)
                async with progress_lock:
                    result.processed_section_count += 1
                result.sections.append(fallback)
                continue

            _sec_id, section, unmatched = item
            unmatched_global.extend(unmatched)
            result.sections.append(section)

    result.unassigned_text = "\n".join(p for p in orphan + unmatched_global if p).strip()

    if orphan:
        result.sections.append(GeneratedSection(
            section_id=UNASSIGNED,
            title="Unassigned Observations",
            text=(
                "[These observations could not be matched to any template paragraph. "
                "Manual review required.]"
            ),
            status="UNASSIGNED",
            unmatched_observations=orphan,
            grounding_passed=False,
        ))

    full = "\n".join(s.text for s in result.sections) + "\n" + result.unassigned_text
    try:
        pii_scrubber.assert_no_pii(full, context="generated report")
    except pii_scrubber.PiiDetectedError:
        logger.warning("Residual PII in generated output; redacting before export.")
        for s in result.sections:
            s.text = pii_scrubber.scrub(s.text).text
            s.unmatched_observations = [
                pii_scrubber.scrub(u).text for u in s.unmatched_observations
            ]
        result.unassigned_text = pii_scrubber.scrub(result.unassigned_text).text

    return result


def generate_report(
    tenant_id: str,
    raw_notes: str,
    *,
    property_type: str = "",
    tenure: str = "",
    interference_level: InterferenceLevel | None = None,
    survey_level: int = 3,
    retrieval_level: RetrievalLevel = "paragraph",
    report_draft_id: str | None = None,
    only_section_ids: list[str] | None = None,
    reference_document_ids: list[str] | None = None,
    strict_uploaded_only: bool = False,
) -> ReportResult:
    """Synchronous entrypoint — delegates to :func:`generate_report_async`."""
    return _run_coroutine_sync(
        generate_report_async(
            tenant_id,
            raw_notes,
            property_type=property_type,
            tenure=tenure,
            interference_level=interference_level,
            survey_level=survey_level,
            retrieval_level=retrieval_level,
            report_draft_id=report_draft_id,
            only_section_ids=only_section_ids,
            reference_document_ids=reference_document_ids,
            strict_uploaded_only=strict_uploaded_only,
        )
    )