backend/startup.py

"""Startup ingest: report template (schema) + standard paragraphs (MASTER RAG).

The operator bundle in ``Master Standard report and paragraphs/`` contains:
  - PDF report template  -> section structure (schema.json)
  - Word standard paras  -> approved wording (MASTER FAISS)

No user upload is required before the first report.
"""

from __future__ import annotations

import logging
import shutil

from backend.config import settings
from backend.domain import template_discoverer
from backend.domain.notes import parser as notes_parser
from backend.domain.rics_level3_schema import PARENT_SECTION_COUNT
from backend.ingest import pipeline as ingest
from backend.pii import scrubber as pii_scrubber
from backend.rag.index_guard import ensure_reference_indices_clean
from backend.rag.store import get_rag_store
from backend.rag.types import TIER_MASTER, TIER_REFERENCE
from backend.storage import tenant_store
from backend.utils.runtime_paths import ensure_data_drive_runtime_dirs

logger = logging.getLogger(__name__)


def _scrub_stale_faiss_write_artifacts() -> None:
    """Remove half-written FAISS/meta files left by interrupted persists."""
    tenants_root = settings.data_dir_path / "tenants"
    if not tenants_root.is_dir():
        return
    for pattern in (
        "*.write.faiss",
        "*.write.json",
        "index.faiss.bad",
        "meta.json.bad",
    ):
        for path in tenants_root.rglob(pattern):
            try:
                path.unlink(missing_ok=True)
            except OSError as exc:
                logger.debug("Could not remove stale artifact %s: %s", path, exc)


def _load_prebuilt() -> bool:
    """Copy prebuilt schema.json + MASTER FAISS into the tenant store, if set."""
    schema_src = settings.master_template_prebuilt_schema
    faiss_src = settings.master_template_prebuilt_faiss
    if not (schema_src and faiss_src):
        return False
    schema_path = settings.resolve_path(schema_src)
    faiss_path = settings.resolve_path(faiss_src)
    if not (schema_path.is_file() and faiss_path.is_dir()):
        logger.warning(
            "Prebuilt artifacts configured but missing; falling back to discovery."
        )
        return False

    tenant = settings.default_tenant_id
    shutil.copyfile(schema_path, tenant_store.schema_path(tenant))
    dest = tenant_store.faiss_dir(tenant, TIER_MASTER)
    for item in faiss_path.iterdir():
        shutil.copy2(item, dest / item.name)
    logger.info("Loaded prebuilt master artifacts for tenant=%s", tenant)
    return True


def run_startup_ingest() -> dict:
    """Ingest the operator bundle. Returns a summary for the health endpoint."""
    ensure_data_drive_runtime_dirs()
    _scrub_stale_faiss_write_artifacts()

    tenant = settings.default_tenant_id
    summary: dict = {
        "master_loaded": False,
        "sections": 0,
        "paragraph_chunks": 0,
        "report_template": settings.report_template_filename,
        "standard_paragraphs": settings.standard_paragraphs_filename,
        "reference_documents": 0,
    }

    if not settings.master_template_auto_ingest:
        logger.info(
            "master_template_auto_ingest disabled; reference-only generation. "
            "Purging any persisted MASTER tier for tenant=%s.",
            tenant,
        )
        template_discoverer.ensure_canonical_schema(tenant)
        # Reference-only model: make removal real and idempotent so stale operator
        # boilerplate from a prior master-enabled run cannot linger in the index.
        get_rag_store().clear_tier(tenant, TIER_MASTER)
        index_guard = ensure_reference_indices_clean()
        summary["reference_index_guard"] = index_guard
        summary["master_loaded"] = True  # schema (canonical) is present
        summary["sections"] = PARENT_SECTION_COUNT
        summary["section_anchor_vectors"] = notes_parser.initialize_section_anchors()

        if settings.reference_auto_ingest_enabled and not index_guard.get("rebuilt"):
            try:
                ref = ingest.auto_ingest_reference_dir(tenant)
                summary["reference_documents"] = ref["documents"]
            except Exception as exc:  # noqa: BLE001
                logger.warning("Reference auto-ingest failed: %s", exc)
        return summary

    loaded = False
    if _load_prebuilt():
        loaded = True
    else:
        try:
            res = ingest.ingest_operator_bundle(tenant)
            summary["sections"] = res["sections"]
            summary["paragraph_chunks"] = res["paragraph_chunks"]
            loaded = True
            logger.info(
                "Operator bundle ingested: schema from %s (%d sections), "
                "paragraphs from %s (%d chunks)",
                res["report_template"],
                res["sections"],
                res["standard_paragraphs"],
                res["paragraph_chunks"],
            )
        except pii_scrubber.PiiDetectedError as exc:
            logger.critical(
                "STANDARD PARAGRAPHS REJECTED — contains property-identifying PII: %s",
                exc,
            )
        except FileNotFoundError as exc:
            logger.error("Operator bundle file missing: %s", exc)
        except Exception as exc:  # noqa: BLE001
            logger.exception("Operator bundle ingest failed: %s", exc)

    template_discoverer.ensure_canonical_schema(tenant)
    index_guard = ensure_reference_indices_clean()
    summary["reference_index_guard"] = index_guard
    summary["master_loaded"] = (
        loaded or template_discoverer.load_schema(tenant) is not None
    )
    summary["sections"] = PARENT_SECTION_COUNT

    if settings.reference_auto_ingest_enabled and not index_guard.get("rebuilt"):
        try:
            ref = ingest.auto_ingest_reference_dir(tenant)
            summary["reference_documents"] = ref["documents"]
        except Exception as exc:  # noqa: BLE001
            logger.warning("Reference auto-ingest failed: %s", exc)
    elif index_guard.get("rebuilt"):
        summary["reference_documents"] = sum(
            row.get("auto_ingest_docs", 0) + row.get("library_reingested", 0)
            for row in index_guard.get("rebuilt", [])
        )

    anchor_count = notes_parser.initialize_section_anchors()
    summary["section_anchor_vectors"] = anchor_count

    store = get_rag_store()
    logger.info(
        "Startup complete: loaded=%s sections=%d paragraph_chunks=%d reference_chunks=%d anchors=%d",
        summary["master_loaded"],
        summary["sections"],
        store.count(tenant, TIER_MASTER),
        store.count(tenant, TIER_REFERENCE),
        anchor_count,
    )
    return summary