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