"""Agentic RICS inspection pipeline. These are *software agents* (modular components), not Cursor subagents. The HeadAgent orchestrates retrieval + analysis steps and uses the existing LLM adapter to draft final report sections in a consistent RICS tone. Capabilities are defined in :mod:`app.agentic.tools`. When :func:`app.agentic.runtime_status.is_openai_inspector_live` is true (non-empty ``OPENAI_API_KEY`` and ``inspector_tool_agent``), :mod:`app.agentic.inspector_loop` runs an OpenAI **tool-calling** loop so the model chooses which retrieval / KB / duplicate-scan tools to invoke before calling ``submit_inspection_section``. Otherwise (tests / no key) the legacy fixed pipeline runs. See ``GET /health`` → ``rics_inspector``. """ from __future__ import annotations import asyncio import json import logging from dataclasses import asdict from typing import Any from app.config import settings from app.generator.postprocess import _L1_PLACEHOLDER, enforce_verify, strip_l1_advice from app.models.schemas import SearchResult, WritingStyleProfile from app.services.generation import _ai_level_to_params # internal mapping from app.services.provenance_enrichment import fetch_doc_filenames from app.templates.registry import get_template, section_order_for_survey from . import tools as agent_tools from .inspector_loop import _verify_risks, run_inspector_tool_loop from .models import ComplianceNote, EvidenceItem, Finding, RiskItem, StructuredReport from .runtime_status import is_openai_inspector_live def _safe_level(survey_level: int | None) -> int: """Coerce ``survey_level`` to a numeric tier, defaulting to L3 on bad input.""" try: return int(survey_level if survey_level is not None else 3) except Exception: # noqa: BLE001 return 3 logger = logging.getLogger(__name__) def _inspector_bundle(rep: StructuredReport) -> dict[str, Any] | None: """Subset of inspector artifacts safe for JSON responses.""" 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"] = list(rep.tool_trace)[-48:] return out or None class DataExtractionAgent: """Retrieve relevant evidence for each report element.""" async def gather_async( self, *, tenant_id: str, primary_document_id: str, section_code: str, bullets: list[str], reference_document_ids: list[str] | None = None, kb_enabled: bool = True, retrieval_level: str = "paragraph", ) -> list[SearchResult]: query = " ".join([b for b in bullets if b.strip()]) refs = list(reference_document_ids or []) out: list[SearchResult] = [] out.extend( await agent_tools.retrieve_tenant_evidence_async( query=query, tenant_id=tenant_id, primary_document_id=primary_document_id, secondary_document_ids=refs, k=max(settings.retrieval_top_k, 12), rerank_top_n=min(6, settings.rerank_top_n + 3), ) ) if kb_enabled: hl = retrieval_level if retrieval_level in ("document", "section", "paragraph") else None out.extend( await agent_tools.retrieve_kb_guidance_async( query=query, k=10, hierarchy_level=hl, rerank_top_n=5, ) ) return agent_tools.dedupe_search_results(out) def gather( self, *, tenant_id: str, primary_document_id: str, section_code: str, bullets: list[str], reference_document_ids: list[str] | None = None, kb_enabled: bool = True, retrieval_level: str = "paragraph", ) -> list[SearchResult]: """Sync retrieval (tests / blocking contexts). Prefer :meth:`gather_async` in async handlers.""" query = " ".join([b for b in bullets if b.strip()]) refs = list(reference_document_ids or []) out: list[SearchResult] = [] out.extend( agent_tools.retrieve_tenant_evidence( query=query, tenant_id=tenant_id, primary_document_id=primary_document_id, secondary_document_ids=refs, k=max(settings.retrieval_top_k, 12), rerank_top_n=min(6, settings.rerank_top_n + 3), ) ) if kb_enabled: hl = retrieval_level if retrieval_level in ("document", "section", "paragraph") else None out.extend( agent_tools.retrieve_kb_guidance( query=query, k=10, hierarchy_level=hl, rerank_top_n=5, ) ) return agent_tools.dedupe_search_results(out) class StandardsComplianceAgent: """Check for typical RICS report hygiene and missing essentials.""" def check( self, *, section_code: str, bullets: list[str], kb_guidance: list[str] | None = None, survey_level: int | None = None, ) -> list[ComplianceNote]: template = get_template(section_code, survey_level) expected = template.expected_fields if template else [] raw = " ".join(bullets).lower() notes: list[ComplianceNote] = [] for f in expected[:10]: # cap to avoid noise ok = f.replace("_", " ") in raw or f.lower() in raw notes.append( ComplianceNote( standard="RICS Home Survey Standard (structure hygiene)", note=f"Expected field '{f}' appears in notes: {'yes' if ok else 'no'}", status="OK" if ok else "Review", ) ) if kb_guidance: notes.append( ComplianceNote( standard="Local KB (RICS/exemplar guidance)", note=f"Retrieved {len(kb_guidance)} guidance snippet(s) relevant to {section_code}.", status="OK", ) ) return notes class RiskAssessmentAgent: """Translate findings into risk items using LLM-based contextual reasoning. Falls back to keyword heuristics when no OpenAI key is configured (e.g. tests). """ _SYSTEM = ( "You are a Chartered Building Surveyor (MRICS) assessing inspection notes.\n" "Return ONLY a JSON array of risk objects. Each object must have exactly these keys:\n" " category (string), risk (string), severity (\"Low\"|\"Medium\"|\"High\"|\"Critical\"),\n" " likelihood (\"Low\"|\"Medium\"|\"High\"), action (string).\n" "Rules:\n" "- Maximum 5 items.\n" "- Base severity ONLY on what the notes explicitly state — do NOT invent defects.\n" "- If notes say \"no signs of X\", \"satisfactory\", or \"good condition\", do NOT flag X.\n" "- If no defects are mentioned, return exactly one item: category \"General\", severity \"Low\", likelihood \"Low\".\n" "Output ONLY the JSON array, no markdown fences, no commentary." ) # Keys the frozen RiskItem dataclass accepts via kwargs. The LLM contract above # asks for exactly these five — but models drift and routinely add extras like # "description", "details", "reasoning", or "evidence". Passing those straight # into RiskItem(**item) raises TypeError, which the broad except below would # silently swallow and downgrade the entire batch to crude keyword heuristics. # We keep `evidence` out of this whitelist on purpose: it's a structural # tuple[EvidenceItem, ...] field that the model can't populate correctly. _RISK_ITEM_FIELDS: frozenset[str] = frozenset( ("category", "risk", "severity", "likelihood", "action") ) @classmethod def _coerce_risk_item(cls, raw: Any) -> RiskItem | None: """Build a RiskItem from one LLM-emitted dict, tolerant of drift. - Unknown keys are dropped (so e.g. an extra ``"description"`` is silently ignored instead of nuking the whole batch). - A single item missing a required key is skipped, not fatal — the rest of the batch survives. - Returns ``None`` if the raw value is unusable. """ if not isinstance(raw, dict): return None clean: dict[str, str] = {} for k, v in raw.items(): if k in cls._RISK_ITEM_FIELDS: clean[k] = "" if v is None else str(v).strip() if not cls._RISK_ITEM_FIELDS.issubset(clean): missing = sorted(cls._RISK_ITEM_FIELDS - set(clean)) logger.warning( "LLM risk item dropped — missing required key(s) %s: %r", missing, raw, ) return None return RiskItem(**clean) async def assess(self, *, section_code: str, bullets: list[str]) -> list[RiskItem]: from app.config import settings if not settings.openai_api_key: return self._keyword_fallback(bullets) notes = "\n".join(f"- {b}" for b in bullets if b.strip()) or "No notes provided." try: from app.llm.openai_chat import chat_completions_create user_content = f"Section: {section_code}\n\nInspection notes:\n{notes}" raw = await chat_completions_create( messages=[ {"role": "system", "content": self._SYSTEM}, {"role": "user", "content": user_content}, ], model=settings.chat_model, max_tokens=600, temperature=0.0, phase="risk_assessment", section_id=section_code, ) items = json.loads(raw) if not isinstance(items, list): raise ValueError("Expected JSON array") parsed = [ item for item in (self._coerce_risk_item(it) for it in items[:5]) if item is not None ] if not parsed: logger.warning( "LLM returned %d risk item(s) but none parsed cleanly — using keyword fallback", len(items), ) return self._keyword_fallback(bullets) return parsed except Exception as exc: logger.warning("LLM risk assessment failed (%s) — using keyword fallback", exc) return self._keyword_fallback(bullets) def _keyword_fallback(self, bullets: list[str]) -> list[RiskItem]: """Simple keyword fallback used when OpenAI is unavailable.""" text = " ".join(bullets).lower() risks: list[RiskItem] = [] def add(cat: str, risk: str, sev: str, lik: str, action: str) -> None: risks.append(RiskItem(category=cat, risk=risk, severity=sev, likelihood=lik, action=action)) if any(k in text for k in ("damp", "mould", "penetrating", "rising")): add("Moisture", "Moisture ingress — investigation required", "Medium", "Medium", "Investigate source; carry out repairs and monitor.") if any(k in text for k in ("crack", "movement", "subsidence", "bulging")): add("Structural", "Structural movement requiring specialist review", "High", "Medium", "Seek structural engineer review before commitment.") if any(k in text for k in ("electrical", "consumer unit", "rcd", "wiring", "fuse")): add("Electrical", "Electrical safety compliance uncertain", "High", "Medium", "Obtain EICR by a qualified electrician.") if any(k in text for k in ("gas", "boiler", "flue", "carbon monoxide")): add("Gas", "Gas safety — appliances require certification", "High", "Low", "Obtain Gas Safe service and flue test.") if not risks: add("General", "No significant defects identified in inspection notes", "Low", "Low", "Maintain property and address minor defects as they arise.") return risks class ReportStructuringAgent: """Convert artifacts to a clean RICS narrative outline.""" def outline( self, *, section_code: str, bullets: list[str], evidence_snippets: list[str], compliance: list[ComplianceNote], risks: list[RiskItem], survey_level: int | None = None, ) -> dict[str, Any]: template = get_template(section_code, survey_level) sk = template.skeleton if template else f"[{section_code}]: [content]." return { "section_code": section_code, "section_title": (template.title if template else section_code), "skeleton": sk, "bullets": bullets, "evidence": evidence_snippets[:10], "compliance": [asdict(x) for x in compliance][:10], "risks": [asdict(x) for x in risks][:10], } def render_report_text( *, title: str, blocks: dict[str, str], section_code: str | None = None, survey_level: int | None = None, ) -> str: """Render report text for one section. For Level 3 element sections (outside/inside/services/grounds), real RICS PDFs use a more fluid narrative rather than repeating fixed subheadings per element. """ code = (section_code or "").strip().upper() lvl = int(survey_level) if survey_level is not None else None # L3 narrative style for core element sections (E/F/G/H): cohesive paragraph(s), no nested subheadings. if lvl == 3 and code and code[0] in ("E", "F", "G", "H") and any(ch.isdigit() for ch in code[1:]): # Prefer condition assessment as the spine, then append risks/recs only if present. parts: list[str] = [] ca = (blocks.get("Condition Assessment") or "").strip() dr = (blocks.get("Defects and Risks") or "").strip() rec = (blocks.get("Recommendations") or "").strip() if ca: parts.append(ca) if dr and (not ca or dr.lower() not in ca.lower()): parts.append(dr) if rec and (rec.lower() not in " ".join(parts).lower()): parts.append(rec) out = "\n\n".join([p for p in parts if p]).strip() return out or (blocks.get("Executive Summary") or "").strip() or title.strip() # Default headed style (kept for A–D, I–L and non-L3 packs). parts2: list[str] = [title.strip()] for h in ("Executive Summary", "Property Description", "Condition Assessment", "Defects and Risks", "Recommendations"): body = (blocks.get(h) or "").strip() if not body: continue parts2.append(f"\n\n{h}\n{body}") return "\n".join(parts2).strip() def _dedupe_risks(items: list[RiskItem]) -> list[RiskItem]: seen: set[tuple[str, str]] = set() out: list[RiskItem] = [] for r in items: k = (str(r.category).strip().lower(), str(r.risk).strip().lower()) if k in seen: continue seen.add(k) out.append(r) return out # Severity ordering MUST match the LLM prompt contract in `RiskAssessmentAgent._SYSTEM`, # which currently allows {Critical, High, Medium, Low}. Lower sort-key = higher priority, # so Critical comes first and any unrecognised value falls to the bottom (instead of being # silently treated as more important than Low — the previous behaviour). Likelihood per # the contract is {Low, Medium, High} only — no Critical there. _SEVERITY_ORDER: dict[str, int] = {"critical": 0, "high": 1, "medium": 2, "low": 3} _LIKELIHOOD_ORDER: dict[str, int] = {"high": 0, "medium": 1, "low": 2} def _risk_priority_key(r: RiskItem) -> tuple[int, int]: sev = _SEVERITY_ORDER.get(str(r.severity).lower(), len(_SEVERITY_ORDER)) lik = _LIKELIHOOD_ORDER.get(str(r.likelihood).lower(), len(_LIKELIHOOD_ORDER)) return sev, lik def render_full_report_text( *, title: str, executive_summary: str, sections: list[tuple[str, str, str]], consolidated_risks: list[RiskItem], recommendations: list[str], ) -> str: """Render a single end-to-end report text with headings.""" parts: list[str] = [title.strip()] if executive_summary.strip(): parts.append(f"\n\nExecutive Summary\n{executive_summary.strip()}") if consolidated_risks: lines = [] for i, r in enumerate(consolidated_risks[:20], 1): lines.append( f"{i}. [{r.category}] {r.risk} (Severity: {r.severity}, Likelihood: {r.likelihood}) — {r.action}" ) parts.append("\n\nDefects and Risks (consolidated)\n" + "\n".join(lines)) if recommendations: rec_lines = [] for i, t in enumerate([x for x in recommendations if x.strip()][:20], 1): rec_lines.append(f"{i}. {t.strip()}") parts.append("\n\nRecommendations (summary)\n" + "\n".join(rec_lines)) # Per-section detail if sections: parts.append("\n\nCondition Assessment (by section)") for code, sec_title, sec_text in sections: body = (sec_text or "").strip() if not body: continue parts.append(f"\n\n{code} — {sec_title}\n{body}") return "\n".join(parts).strip() class HeadAgent: """Professional RICS inspector orchestrator.""" def __init__(self) -> None: self.extractor = DataExtractionAgent() self.compliance = StandardsComplianceAgent() self.risk = RiskAssessmentAgent() self.structurer = ReportStructuringAgent() async def generate_section_report( self, *, db, tenant_id: str, primary_document_id: str, section_code: str, bullets: list[str], style_profile: WritingStyleProfile, ai_percent: int = 50, retrieval_level: str = "paragraph", reference_document_ids: list[str] | None = None, similarity_scan: bool = False, peer_sections: dict[str, str] | None = None, similarity_exclude_document_ids: list[str] | None = None, survey_level: int | None = None, ) -> StructuredReport: if is_openai_inspector_live(): rep, _ = await run_inspector_tool_loop( db=db, tenant_id=tenant_id, primary_document_id=primary_document_id, section_code=section_code, bullets=bullets, style_profile=style_profile, ai_percent=ai_percent, retrieval_level=retrieval_level, reference_document_ids=list(reference_document_ids or []), peer_sections=dict(peer_sections or {}), survey_level=survey_level, ) return rep # Legacy fixed pipeline (mock adapter, or inspector disabled) # 1) gather evidence (tenant + knowledge base) hits = await self.extractor.gather_async( tenant_id=tenant_id, primary_document_id=primary_document_id, section_code=section_code, bullets=bullets, reference_document_ids=reference_document_ids, kb_enabled=True, retrieval_level=retrieval_level, ) # 2) enrich evidence items with filenames (tenant docs) + kb source labels tenant_doc_ids = {r.doc_id for r in hits if r.tenant_id == tenant_id and r.doc_id} filenames = await fetch_doc_filenames(db, tenant_id, tenant_doc_ids) evidence_items: list[EvidenceItem] = [] snippets: list[str] = [] kb_guidance: list[str] = [] for r in hits: is_kb = bool(getattr(r, "kb", False)) or r.tenant_id == settings.knowledge_base_tenant_id src = (getattr(r, "source", None) or None) if is_kb else filenames.get(r.doc_id) if not src and is_kb: src = "Local RICS knowledge base" evidence_items.append( EvidenceItem( doc_id=r.doc_id, chunk_id=r.chunk_id, score=float(r.score), text=r.text, source=src, section_hint=getattr(r, "section_title", None), kb=is_kb, ) ) snippets.append(r.text) if is_kb: kb_guidance.append(r.text) # 3) compliance + risk comp = self.compliance.check( section_code=section_code, bullets=bullets, kb_guidance=kb_guidance[:3], survey_level=survey_level, ) risks = await self.risk.assess(section_code=section_code, bullets=bullets) if similarity_scan and db is not None: query = " ".join([b for b in bullets if b.strip()]) try: sim = await agent_tools.find_similar_library_and_peers( db, tenant_id, text=query or section_code, section_code=section_code, peer_sections=dict(peer_sections or {}), exclude_document_ids=list(similarity_exclude_document_ids or []), ) n_lib = len(sim.library_matches) n_peer = len(sim.draft_overlaps) if n_lib or n_peer: comp = list(comp) + [ ComplianceNote( standard="Corpus hygiene (find_similar_library_and_peers tool)", note=( f"Similarity scan: {n_lib} indexed library match(es), " f"{n_peer} draft overlap(s) with peer sections. " "Reconcile duplicates before sign-off." ), status="Review" if n_peer else "OK", ) ] except Exception: logger.exception("Agent tool find_similar_library_and_peers failed for section=%s", section_code) # 4) structure prompt variables outline = self.structurer.outline( section_code=section_code, bullets=bullets, evidence_snippets=snippets, compliance=comp, risks=risks, survey_level=survey_level, ) from app.llm import generation_facade as gen_llm ai_params = _ai_level_to_params(3, ai_percent=ai_percent) skeleton = outline["skeleton"] doc_ctx = outline["evidence"][:3] para_ctx = outline["evidence"][3:] base = await gen_llm.generate_section( skeleton=skeleton, bullets=bullets, snippets=[], style_profile=style_profile if ai_percent > 5 else None, temperature=float(ai_params["temperature"]), creativity_hint=str(ai_params["creativity_hint"]) + "\n\nAGENTIC CONTEXT: You are drafting as a Chartered Building Surveyor (MRICS). " "Be risk-based and recommendation-led.", document_context=doc_ctx, hierarchy_section_snippets=None, paragraph_snippets=para_ctx, style_anchor=None, survey_level=survey_level, tenant_id=tenant_id, ) # Build a section-scoped structured report (full-report assembly is # handled by API layer). The legacy non-agentic path runs only when # `is_openai_inspector_live()` is false (no key, or the agentic flag # is off — typically tests / offline). It still emits LLM-generated # text via the adapter, so we mirror the same regex non-invention # guard the agentic path applies to its risks. The fast regex pass # catches the common offenders (postcodes, addresses, named persons) # without adding the latency / cost of an LLM grounding round-trip # in a path that's mostly hit when no key is configured anyway. title = outline["section_title"] verified_base = enforce_verify(text=base, bullets=bullets, snippets=snippets) risks = _verify_risks(risks, bullets=bullets, snippets=snippets) # Tier-aware behavioural enforcement (parity with inspector_loop). # The legacy path runs in tests / no-key offline environments — but # also as the real production path when the inspector flag is off. # Either way an L1 product must read as observation, never advice. legacy_lvl = _safe_level(survey_level) if legacy_lvl <= 1: verified_base = strip_l1_advice(verified_base) risks = [ RiskItem( category=r.category, risk=r.risk, severity=r.severity, likelihood=r.likelihood, action=strip_l1_advice(r.action) if r.action else r.action, evidence=r.evidence, ) for r in risks ] defects = " ".join( f"{r.category}: {r.risk} ({r.severity}/{r.likelihood})." for r in risks ).strip()[:2400] recs = _L1_PLACEHOLDER else: defects = " ".join( f"{r.category}: {r.risk} ({r.severity}/{r.likelihood}). {r.action}" for r in risks ).strip()[:2400] recs = " ".join(r.action for r in risks).strip()[:1600] # Previous behaviour set both `property_description` and # `condition_assessment` to the same `base[:900]`, plus a third # `executive_summary = title: base` containing yet another copy of # the text. That's three views of the same truncated string — # exactly the "summary feel" the user reported. Place the full # verified draft once, in `condition_assessment`, and leave # `property_description` empty so the renderer skips it rather than # repeating itself. `executive_summary` becomes a short MRICS-style # headline rather than a third copy of the body. return StructuredReport( executive_summary=f"{title} — Chartered Building Surveyor inspection summary.", property_description="", condition_assessment=verified_base.strip(), defects_and_risks=defects, recommendations=recs, findings=(), risks=tuple(risks), compliance=tuple(comp), evidence_items=tuple(evidence_items), tool_trace=(), ) async def generate_full_report( *, db, tenant_id: str, primary_document_id: str, bullets_by_section: dict[str, list[str]], style_profile: WritingStyleProfile, ai_percent: int = 50, retrieval_level: str = "paragraph", reference_document_ids: list[str] | None = None, similarity_scan: bool = False, peer_sections: dict[str, str] | None = None, similarity_exclude_document_ids: list[str] | None = None, survey_level: int | None = None, ) -> dict[str, Any]: """Generate a full multi-section report as a JSON structure + stitched report text.""" head = HeadAgent() out: dict[str, Any] = {"sections": {}} all_risks: list[RiskItem] = [] section_detail_for_stitch: list[tuple[str, str, str]] = [] recs: list[str] = [] codes = section_order_for_survey(survey_level) def _missing_placeholder(sec_code: str) -> tuple[str, str]: template = get_template(sec_code, survey_level) sec_title_local = template.title if template else sec_code if template and template.has_condition_rating: missing = "Not inspected. Condition Rating NI." else: missing = "Not applicable." return sec_title_local, missing from app.db.database import effective_section_concurrency # Bounded concurrent section generation. Decoupled from enable_async_pipeline: # section work is dominated by I/O-bound inspector/LLM calls, so concurrency # cuts wall-clock time. The semaphore bounds how many sections hold a DB # session + inspector loop at once; the global LLM throttle still caps # in-flight provider calls. _section_concurrency = effective_section_concurrency() if _section_concurrency > 1: import structlog log = structlog.get_logger(__name__) _section_sem = asyncio.Semaphore(_section_concurrency) tasks: list[tuple[str, asyncio.Task[Any]]] = [] for code in codes: bullets = bullets_by_section.get(code) or [] if not bullets: sec_title, missing = _missing_placeholder(code) out["sections"][code] = { "executive_summary": missing, "property_description": missing, "condition_assessment": missing, "defects_and_risks": missing, "recommendations": missing, "report_text": render_report_text( title=f"RICS Inspection Report — Section {code}", blocks={ "Executive Summary": missing, "Property Description": missing, "Condition Assessment": missing, "Defects and Risks": missing, "Recommendations": missing, }, section_code=code, survey_level=survey_level, ), "risks": [], "compliance": [], "evidence_items": [], "tool_trace": [], "inspector": None, } section_detail_for_stitch.append((code, sec_title, missing)) continue async def _run_one_section( sec_code: str, sec_bullets: list[str], ) -> Any: async with _section_sem: from app.db.database import get_session_factory factory = get_session_factory() async with factory() as section_db: return await head.generate_section_report( db=section_db, tenant_id=tenant_id, primary_document_id=primary_document_id, section_code=sec_code, bullets=sec_bullets, 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=survey_level, ) tasks.append( ( code, asyncio.create_task(_run_one_section(code, bullets)), ) ) if tasks: results = await asyncio.gather( *(t for _, t in tasks), return_exceptions=True ) for (code, _task), rep in zip(tasks, results): if isinstance(rep, BaseException): log.error( "section_generation_failed", event="section_generation_failed", phase="generate_full_report", section_id=code, cache_hit=None, exc_type=type(rep).__name__, error=str(rep), ) sec_title, missing = _missing_placeholder(code) missing_detail = "Section generation failed; verify inputs and regenerate." out["sections"][code] = { "executive_summary": missing_detail, "property_description": missing_detail, "condition_assessment": missing_detail, "defects_and_risks": missing_detail, "recommendations": missing_detail, "report_text": render_report_text( title=f"RICS Inspection Report — Section {code}", blocks={ "Executive Summary": missing_detail, "Property Description": missing_detail, "Condition Assessment": missing_detail, "Defects and Risks": missing_detail, "Recommendations": missing_detail, }, section_code=code, survey_level=survey_level, ), "risks": [], "compliance": [], "evidence_items": [], "tool_trace": [], "inspector": None, } section_detail_for_stitch.append( (code, sec_title, missing) ) continue inspector = _inspector_bundle(rep) out["sections"][code] = { "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, "report_text": render_report_text( title=f"RICS Inspection Report — Section {code}", 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=code, survey_level=survey_level, ), "risks": [asdict(r) for r in rep.risks], "compliance": [asdict(c) for c in rep.compliance], "evidence_items": [asdict(e) for e in rep.evidence_items], "tool_trace": list(rep.tool_trace), "inspector": inspector, } all_risks.extend(list(rep.risks)) template = get_template(code, survey_level) sec_title = template.title if template else code section_detail_for_stitch.append( ( code, sec_title, rep.condition_assessment or rep.property_description or "", ) ) if rep.recommendations: recs.append(rep.recommendations) else: # Legacy sequential generation. for code in codes: bullets = bullets_by_section.get(code) or [] if not bullets: sec_title, missing = _missing_placeholder(code) out["sections"][code] = { "executive_summary": missing, "property_description": missing, "condition_assessment": missing, "defects_and_risks": missing, "recommendations": missing, "report_text": render_report_text( title=f"RICS Inspection Report — Section {code}", blocks={ "Executive Summary": missing, "Property Description": missing, "Condition Assessment": missing, "Defects and Risks": missing, "Recommendations": missing, }, section_code=code, survey_level=survey_level, ), "risks": [], "compliance": [], "evidence_items": [], "tool_trace": [], "inspector": None, } section_detail_for_stitch.append((code, sec_title, missing)) continue rep = await head.generate_section_report( db=db, tenant_id=tenant_id, primary_document_id=primary_document_id, section_code=code, bullets=bullets, 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=survey_level, ) inspector = _inspector_bundle(rep) out["sections"][code] = { "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, "report_text": render_report_text( title=f"RICS Inspection Report — Section {code}", 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=code, survey_level=survey_level, ), "risks": [asdict(r) for r in rep.risks], "compliance": [asdict(c) for c in rep.compliance], "evidence_items": [asdict(e) for e in rep.evidence_items], "tool_trace": list(rep.tool_trace), "inspector": inspector, } all_risks.extend(list(rep.risks)) template = get_template(code, survey_level) sec_title = template.title if template else code section_detail_for_stitch.append( ( code, sec_title, rep.condition_assessment or rep.property_description or "", ) ) if rep.recommendations: recs.append(rep.recommendations) # Consolidate risks across all included sections consolidated = _dedupe_risks(all_risks) consolidated.sort(key=_risk_priority_key) # Build a unified executive summary using the existing adapter (keeps UK English + style). # Tier resolution must happen BEFORE the try/except so the fallback path # can pick a tier-appropriate hardcoded summary; previously `exec_lvl` # was set inside the try block, so the except branch defaulted to a # single string that contained "recommended next steps" — directive # advice phrasing that bypasses the L1 sanitiser and leaks into L1 # products on any adapter failure. exec_lvl = _safe_level(survey_level) try: from app.llm import generation_facade as gen_llm ai_params = _ai_level_to_params(3, ai_percent=ai_percent) # Bullets for the exec summary: top risks + scope summary scope_bullets = [ f"Sections covered: {', '.join(sorted(out['sections'].keys()))}", f"Top risks: {', '.join([f'{r.category} ({r.severity})' for r in consolidated[:5]])}" if consolidated else "Top risks: none highlighted", ] # Tier-aware exec summary length and behaviour. Without # ``survey_level=`` the adapter defaults to L3 word/token budgets # for an L1 report, producing a 600-word executive summary at the # top of an L1 product whose body sections cap at ~90 words — # an obvious tier mismatch the user could see at a glance. if exec_lvl <= 1: length_hint = ( "Write 50–110 words. Level 1 (Condition Report) — observation only; " "do NOT use directive phrasing such as 'we recommend' or 'should be replaced'. " ) elif exec_lvl == 2: length_hint = ( "Write 100–180 words. Level 2 (HomeBuyer) — proportionate buyer-focused " "summary; flag practical next steps for material risks. " ) else: length_hint = ( "Write 140–240 words. Level 3 (Building Survey) — diagnostic summary; " "highlight cause/implication/options for material defects. " ) exec_text = await gen_llm.generate_section( skeleton="Executive Summary: [overall_opinion]. [key_risks]. [next_steps].", bullets=scope_bullets, snippets=[], style_profile=style_profile if ai_percent > 5 else None, temperature=float(ai_params["temperature"]), creativity_hint=str(ai_params["creativity_hint"]) + "\n\n" + length_hint + "Do not invent facts. " + "If something is missing or cannot be verified, omit that unsupported claim " + "instead of writing placeholder text.", document_context=[], hierarchy_section_snippets=None, paragraph_snippets=[r.risk + " — " + r.action for r in consolidated[:8]], style_anchor=None, survey_level=survey_level, tenant_id=tenant_id, ) except Exception: # noqa: BLE001 # Tier-aware hardcoded fallback. The previous single-string fallback # contained "recommended next steps" — `_L1_ADVICE_KEYWORDS_RE` flags # "recommended" as forbidden L1 phrasing. Branch by tier so the L1 # fallback never carries directive language. The L1 variant is # observation-only by construction; L2/L3 variants reference next # steps because their products legitimately include advice. if exec_lvl <= 1: exec_text = ( "This Level 1 Condition Report summarises the inspection observations and " "condition ratings recorded for each element." ) elif exec_lvl == 2: exec_text = ( "This Level 2 Home Survey summarises the available inspection notes and " "supporting evidence. Key risks and proportionate next steps are highlighted " "below." ) else: exec_text = ( "This Level 3 Building Survey summarises the available inspection notes and " "supporting evidence. Key risks and recommended next steps are highlighted " "below." ) # L1 sanitiser on the unified exec summary — applied OUTSIDE the # try/except so it runs on whichever path produced ``exec_text``. A # directive that leaks past the prompt-level guidance is the same # problem here as in the per-section path; previously this only ran # in the success branch, leaving the exception fallback to ship # advice phrasing on adapter failures. if exec_lvl <= 1: exec_text = strip_l1_advice(exec_text) out["consolidated_risks"] = [asdict(r) for r in consolidated] out["executive_summary"] = exec_text out["full_report_text"] = render_full_report_text( title="RICS Inspection Report (agentic draft)", executive_summary=exec_text, sections=section_detail_for_stitch, consolidated_risks=consolidated, recommendations=recs, ) return out