satquery_agent/reporting.py

"""Bounded SatQuery result caching and local mission-report generation."""

from __future__ import annotations

import copy
import csv
import hashlib
import io
import json
import os
import re
import tempfile
import threading
import time
import zipfile
from collections import OrderedDict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple

from PIL import Image, ImageDraw, ImageFont

from .image_ingestion import preview_file_path, remove_preview_url
from .evidence_lifecycle import normalize_translation_evidence
from .models import (
    AgentResponse,
    CacheMetadata,
    ReportArtifact,
    ReportFormat,
    ReportResponse,
    ResponseStatus,
    TaskType,
)


SCHEMA_VERSION = "1.0"
TOOL_VERSION = "satquery-sar-routing-1.2-pix2pix-evidence"
DEFAULT_MAX_RESULTS = 32
DEFAULT_TTL_SECONDS = 30 * 60
REPORT_DIR = Path(
    os.getenv("SATQUERY_REPORT_DIR", str(Path(tempfile.gettempdir()) / "geovision-satquery-reports"))
).resolve()
REPORT_DIR.mkdir(parents=True, exist_ok=True)
_PREVIEW_PATTERN = re.compile(r"^/api/agent/previews/([0-9a-f]{32}\.png)$")
_ARTIFACT_PATTERN = re.compile(r"^GeoVision_SatQuery_[a-z_]+_[0-9a-f-]{36}\.(pdf|json|csv|zip)$")


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def _configured_int(name: str, default: int, minimum: int) -> int:
    try:
        return max(minimum, int(os.getenv(name, str(default))))
    except ValueError:
        return default


def cache_key_for(
    *,
    primary_hash: str,
    secondary_hash: Optional[str],
    task: TaskType,
    normalized_query: str,
    safe_parameters: Dict[str, Any],
) -> str:
    payload = {
        "primary_hash": primary_hash,
        "secondary_hash": secondary_hash,
        "task": task.value,
        "query": " ".join(normalized_query.lower().split()),
        "tool_version": TOOL_VERSION,
        "parameters": safe_parameters,
    }
    return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()


def _preview_urls(value: Any) -> List[str]:
    output: List[str] = []
    if isinstance(value, dict):
        for nested in value.values():
            output.extend(_preview_urls(nested))
    elif isinstance(value, list):
        for nested in value:
            output.extend(_preview_urls(nested))
    elif isinstance(value, str) and _PREVIEW_PATTERN.fullmatch(value):
        output.append(value)
    return list(dict.fromkeys(output))


@dataclass
class MissionRecord:
    response: AgentResponse
    cache_key: str
    generated_at: str
    touched_at: float


class MissionResultStore:
    """Process-local bounded store. Restarting the API intentionally clears it."""

    def __init__(self) -> None:
        self.max_items = _configured_int("SATQUERY_RESULT_CACHE_MAX_ITEMS", DEFAULT_MAX_RESULTS, 1)
        self.ttl_seconds = _configured_int("SATQUERY_RESULT_CACHE_TTL_SECONDS", DEFAULT_TTL_SECONDS, 60)
        self._records: "OrderedDict[str, MissionRecord]" = OrderedDict()
        self._cache_index: Dict[str, str] = {}
        self._lock = threading.RLock()
        self._hits = 0
        self._misses = 0

    def _discard(self, request_id: str) -> None:
        record = self._records.pop(request_id, None)
        if record is None:
            return
        if self._cache_index.get(record.cache_key) == request_id:
            self._cache_index.pop(record.cache_key, None)
        for url in _preview_urls(record.response.model_dump(mode="json")):
            remove_preview_url(url)

    def _cleanup(self) -> None:
        cutoff = time.time() - self.ttl_seconds
        for request_id, record in list(self._records.items()):
            if record.touched_at < cutoff:
                self._discard(request_id)
        while len(self._records) > self.max_items:
            self._discard(next(iter(self._records)))

    def put(self, response: AgentResponse, cache_key: str, generated_at: Optional[str] = None) -> None:
        with self._lock:
            self._cleanup()
            response = normalize_translation_evidence(response)
            request_id = response.request_id
            self._records[request_id] = MissionRecord(
                response=response.model_copy(deep=True),
                cache_key=cache_key,
                generated_at=generated_at or utc_now(),
                touched_at=time.time(),
            )
            self._records.move_to_end(request_id)
            self._cache_index[cache_key] = request_id
            self._cleanup()

    def get_by_request(self, request_id: str) -> Optional[MissionRecord]:
        with self._lock:
            self._cleanup()
            record = self._records.get(request_id)
            if record is None:
                return None
            record.touched_at = time.time()
            self._records.move_to_end(request_id)
            return copy.deepcopy(record)

    def get_by_cache(self, cache_key: str) -> Optional[MissionRecord]:
        with self._lock:
            self._cleanup()
            request_id = self._cache_index.get(cache_key)
            if request_id is None:
                self._misses += 1
                return None
            record = self.get_by_request(request_id)
            if record is None:
                self._misses += 1
                return None
            self._hits += 1
            return record

    def snapshot(self) -> Dict[str, int]:
        with self._lock:
            self._cleanup()
            return {
                "stored_results": len(self._records),
                "max_results": self.max_items,
                "ttl_seconds": self.ttl_seconds,
                "hits": self._hits,
                "misses": self._misses,
            }

    def clear(self) -> None:
        with self._lock:
            for request_id in list(self._records):
                self._discard(request_id)
            self._hits = 0
            self._misses = 0


MISSION_STORE = MissionResultStore()


def mark_fresh(response: AgentResponse, cache_key: str, generated_at: Optional[str] = None) -> AgentResponse:
    timestamp = generated_at or utc_now()
    return response.model_copy(
        update={
            "cache": CacheMetadata(
                cached=False,
                original_generation_timestamp=timestamp,
                retrieval_timestamp=timestamp,
                tool_version=TOOL_VERSION,
                cache_key_prefix=cache_key[:12],
            )
        }
    )


def cached_response(record: MissionRecord) -> AgentResponse:
    response = normalize_translation_evidence(record.response.model_copy(deep=True))
    return response.model_copy(
        update={
            "cache": CacheMetadata(
                cached=True,
                original_generation_timestamp=record.generated_at,
                retrieval_timestamp=utc_now(),
                tool_version=TOOL_VERSION,
                cache_key_prefix=record.cache_key[:12],
            )
        }
    )


def result_is_reportable(response: AgentResponse) -> bool:
    return response.status in {ResponseStatus.SUCCESS, ResponseStatus.PARTIAL, ResponseStatus.ALIGNMENT_REQUIRED}


def _sanitize(value: Any, key: str = "") -> Any:
    lowered = key.lower()
    if any(token in lowered for token in ("api_key", "secret", "password", "token", "local_path", "cache_path")):
        return None
    if isinstance(value, dict):
        return {name: _sanitize(item, name) for name, item in value.items() if _sanitize(item, name) is not None}
    if isinstance(value, list):
        return [_sanitize(item) for item in value]
    if isinstance(value, str):
        if value.startswith(("/Users/", "/home/", "/root/", "C:\\")):
            return "[redacted local path]"
    return value


def _named_previews(value: Any, prefix: str = "evidence") -> List[Tuple[str, str]]:
    output: List[Tuple[str, str]] = []
    if isinstance(value, dict):
        for name, nested in value.items():
            output.extend(_named_previews(nested, f"{prefix}.{name}"))
    elif isinstance(value, list):
        for index, nested in enumerate(value):
            output.extend(_named_previews(nested, f"{prefix}.{index + 1}"))
    elif isinstance(value, str) and _PREVIEW_PATTERN.fullmatch(value):
        output.append((prefix.replace("_", " ").replace(".", " · ").title(), value))
    seen = set()
    return [(label, url) for label, url in output if not (url in seen or seen.add(url))]


def _report_previews(response: Dict[str, Any]) -> List[Tuple[str, str]]:
    preferred: List[Tuple[str, str]] = []
    grounding = response.get("grounding_result") or {}
    sar_water = response.get("sar_water_analysis") or {}
    sar_scene = response.get("sar_scene_analysis") or {}
    grounding_preview = grounding.get("annotated_preview_url")
    if isinstance(grounding_preview, str) and _PREVIEW_PATTERN.fullmatch(grounding_preview):
        preferred.append(("Grounding · Model-produced boxes and scores", grounding_preview))
    change = response.get("change_analysis") or {}
    for name, url in (change.get("previews") or {}).items():
        if isinstance(url, str) and _PREVIEW_PATTERN.fullmatch(url):
            preferred.append((f"Change · {name.replace('_', ' ').title()}", url))
    cross = response.get("cross_modal_analysis") or {}
    for product in cross.get("evidence_products") or []:
        url = product.get("reference")
        if isinstance(url, str) and _PREVIEW_PATTERN.fullmatch(url):
            preferred.append((product["label"], url))
    for name, url in (cross.get("previews") or {}).items():
        if isinstance(url, str) and _PREVIEW_PATTERN.fullmatch(url):
            preferred.append((f"Optical–SAR · {name.replace('_', ' ').title()}", url))
    single = ((response.get("vqa_details") or {}).get("single_image_evidence") or {})
    for name, url in (single.get("previews") or {}).items():
        if isinstance(url, str) and _PREVIEW_PATTERN.fullmatch(url):
            preferred.append((f"Single-image · {name.replace('_', ' ').title()}", url))
    for label, metadata in (("Primary input preview", response.get("primary_image_metadata")), ("Secondary input preview", response.get("secondary_image_metadata"))):
        url = (metadata or {}).get("preview_url")
        if isinstance(url, str) and _PREVIEW_PATTERN.fullmatch(url):
            role = (metadata or {}).get("observation_role")
            preferred.append((f"{role.title()} source" if role else label, url))
    for product in response.get("evidence") or []:
        url = product.get("reference") if isinstance(product, dict) else None
        if isinstance(url, str) and _PREVIEW_PATTERN.fullmatch(url):
            preferred.append((product.get("label") or "Published evidence", url))
    fallback_response = copy.deepcopy(response)
    fallback_response.pop("sar_translated_optical_analysis", None)
    fallback_response.pop("evidence", None)
    seen = set()
    return [(label, url) for label, url in preferred + _named_previews(fallback_response) if not (url in seen or seen.add(url))]


def _statistics(response: Dict[str, Any]) -> Dict[str, Any]:
    details = response.get("vqa_details") or {}
    single = details.get("single_image_evidence") or {}
    change = response.get("change_analysis") or {}
    cross = response.get("cross_modal_analysis") or {}
    grounding = response.get("grounding_result") or {}
    sar_water = response.get("sar_water_analysis") or {}
    sar_scene = response.get("sar_scene_analysis") or {}
    sve = (
        response.get("sve_result")
        or (change.get("sve_result") if isinstance(change, dict) else None)
        or (cross.get("sve_result") if isinstance(cross, dict) else None)
        or {}
    )
    detections = grounding.get("detections") or []
    rejected = grounding.get("rejected_candidates") or []
    return {
        "controlled_answer_statistics": details.get("statistics_used") or {},
        "single_image_evidence": single.get("statistics") or {},
        "change_statistics": change.get("statistics") or {},
        "semantic_change_summary": response.get("semantic_change_summary") or change.get("semantic_change_summary") or {},
        "deterministic_change_statistics": change.get("deterministic_statistics") or {},
        "ttp_change_statistics": response.get("ttp_result") or change.get("ttp_result") or {},
        "mask_comparison": response.get("mask_comparison") or change.get("mask_comparison") or {},
        "cross_modal_statistics": cross.get("statistics") or {},
        "grounding_statistics": {
            "target_phrase": grounding.get("target_phrase"),
            "detection_count": len(detections),
            "accepted_detection_count": len(detections),
            "rejected_candidate_count": len(rejected),
            "top_alignment_score": detections[0].get("score") if detections else None,
            "highest_rejected_alignment_score": max(
                (candidate.get("score") for candidate in rejected if candidate.get("score") is not None),
                default=None,
            ),
            "quality_policy": grounding.get("quality_policy"),
            "mask_refinement_connected": False,
        } if grounding else {},
        "sar_water_statistics": {name: sar_water.get(name) for name in ("water_detected", "image_area_percent", "geographic_area_square_meters", "geographic_area_method", "candidate_pixels", "valid_pixels", "heuristic_reliability", "input_quality_score", "threshold")} if sar_water else {},
        "sar_scene_statistics": {name: sar_scene.get(name) for name in ("valid_pixel_percent", "low_backscatter_percent", "mid_backscatter_percent", "high_backscatter_percent", "texture_index", "input_quality_score")} if sar_scene else {},
        "sve_scene_evidence": {
            "status": sve.get("status"),
            "device": sve.get("device"),
            "runtime_ms": sve.get("runtime_ms"),
            "top_scene_priors": sve.get("scene_priors") or [],
            "caption_consistency": sve.get("caption_consistency"),
            "vqa_consistency": sve.get("vqa_consistency"),
            "grounding_support": sve.get("grounding_support"),
            "semantic_comparison": sve.get("semantic_comparison"),
        } if sve else {},
    }


def build_report_document(response: AgentResponse, generated_at: str, artifact_names: Iterable[str]) -> Dict[str, Any]:
    safe_response = _sanitize(normalize_translation_evidence(response).model_dump(mode="json"))
    details = safe_response.get("vqa_details") or {}
    grounding = safe_response.get("grounding_result") or {}
    sve = (
        safe_response.get("sve_result")
        or (safe_response.get("change_analysis") or {}).get("sve_result")
        or (safe_response.get("cross_modal_analysis") or {}).get("sve_result")
    )
    report = {
        "schema_version": SCHEMA_VERSION,
        "report": {
            "title": "GeoVision SatQuery Mission Report",
            "analysis_type": safe_response.get("task"),
            "request_id": safe_response.get("request_id"),
            "generated_timestamp": generated_at,
            "status": safe_response.get("status"),
            "result_status": safe_response.get("result_status"),
        },
        "input_summary": {
            "primary": safe_response.get("primary_image_metadata"),
            "secondary": safe_response.get("secondary_image_metadata"),
            "compatibility": safe_response.get("pair_compatibility"),
            "observations_by_role": {item["observation_role"]: item for item in (safe_response.get("primary_image_metadata"), safe_response.get("secondary_image_metadata")) if item and item.get("observation_role")},
            "evidence_products": (safe_response.get("cross_modal_analysis") or {}).get("evidence_products") or [],
        },
        "user_query": {
            "original_query": details.get("original_question") or grounding.get("original_query") or (safe_response.get("execution") or {}).get("permitted_parameters", {}).get("query"),
            "detected_task": safe_response.get("task"),
            "question_category": details.get("question_category"),
            "target_concept": details.get("target_concept") or grounding.get("target_phrase"),
        },
        "answer": {
            "text": safe_response.get("answer"),
            "source": details.get("answer_source") or ((safe_response.get("model") or {}).get("tool_id")),
            "status": safe_response.get("status"),
            "result_status": safe_response.get("result_status"),
            "warnings": safe_response.get("warnings") or [],
        },
        "evidence": {
            "items": safe_response.get("evidence") or [],
            "preview_products": [{"label": label, "url": url} for label, url in _report_previews(safe_response)],
            "disclaimer": "Evidence products are model-generated or deterministic heuristic products as labelled; none are ground truth.",
        },
        "statistics": _statistics(safe_response),
        "confidence": safe_response.get("confidence"),
        "specialist_provenance": {
            "selected_tools": (safe_response.get("execution") or {}).get("selected_tools") or [],
            "model": safe_response.get("model"),
            "caption_details": safe_response.get("caption_details"),
            "grounding": {
                "device": grounding.get("device"),
                "input": grounding.get("input"),
                "model_load_ms": grounding.get("model_load_ms"),
                "model_reused": grounding.get("model_reused"),
                "quality_policy": grounding.get("quality_policy"),
                "operational_threshold_disclaimer": grounding.get("operational_threshold_disclaimer"),
            } if grounding else None,
            "vqa_method": details.get("method"),
            "cross_modal_method": (safe_response.get("cross_modal_analysis") or {}).get("method"),
            "sar_water_method": (safe_response.get("sar_water_analysis") or {}).get("method"),
            "sar_water_method_version": (safe_response.get("sar_water_analysis") or {}).get("method_version"),
            "sar_preprocessing": ((safe_response.get("sar_water_analysis") or {}).get("preprocessing") or (safe_response.get("sar_scene_analysis") or {}).get("preprocessing")),
            "sar_translation": safe_response.get("sar_translated_optical_analysis"),
            "heuristic_reliability": (safe_response.get("sar_water_analysis") or {}).get("heuristic_reliability"),
            "model_confidence": (safe_response.get("sar_water_analysis") or {}).get("model_confidence"),
            "change_engine": safe_response.get("change_engine") or (safe_response.get("change_analysis") or {}).get("change_engine"),
            "ttp": safe_response.get("ttp_result") or (safe_response.get("change_analysis") or {}).get("ttp_result"),
            "mask_comparison": safe_response.get("mask_comparison") or (safe_response.get("change_analysis") or {}).get("mask_comparison"),
            "evidence_consistency": safe_response.get("evidence_consistency") or (safe_response.get("change_analysis") or {}).get("evidence_consistency"),
            "semantic_change_summary": safe_response.get("semantic_change_summary") or (safe_response.get("change_analysis") or {}).get("semantic_change_summary"),
            "runtime_ms": (safe_response.get("execution") or {}).get("duration_ms"),
            "satquery_vision_encoder": sve,
        },
        "execution_trace": safe_response.get("execution"),
        "scientific_limitations": list(dict.fromkeys(
            (details.get("limitations") or [])
            + ((safe_response.get("caption_details") or {}).get("limitations") or [])
            + (grounding.get("limitations") or [])
            + (((safe_response.get("cross_modal_analysis") or {}).get("method") or {}).get("limitations") or [])
            + ((safe_response.get("sar_water_analysis") or {}).get("limitations") or [])
            + ((safe_response.get("sar_scene_analysis") or {}).get("limitations") or [])
            + (((safe_response.get("ttp_result") or (safe_response.get("change_analysis") or {}).get("ttp_result") or {}).get("limitations")) or [])
            + ((sve or {}).get("limitations") or [])
            + [
                "Outputs are not ground truth and must be reviewed with source imagery and mission context.",
                "Learned change-detector output is a model-generated binary prediction and is not ground truth.",
                "No hidden geolocation, unsupported causal interpretation, or calibrated semantic probability is produced.",
                "SatQuery Vision Encoder v1 provides scene-level embedding evidence. It does not produce pixel-level segmentation, object grounding, calibrated probabilities, or ground truth.",
            ]
        )),
        "reproducibility": {
            "request_id": safe_response.get("request_id"),
            "tool_version": TOOL_VERSION,
            "method_configuration": (safe_response.get("execution") or {}).get("permitted_parameters") or {},
            "generated_artifact_list": list(artifact_names),
            "cache": safe_response.get("cache"),
            "restart_behavior": "Result and artifact caches are bounded, process-local, and cleared when the backend restarts.",
        },
        "authoritative_response": safe_response,
    }
    return report


def _csv_bytes(document: Dict[str, Any]) -> bytes:
    stream = io.StringIO()
    writer = csv.writer(stream)
    writer.writerow(["section", "metric", "value", "region_id", "region_type", "area_pixels", "bbox"])
    for section, values in document["statistics"].items():
        if not isinstance(values, dict):
            continue
        for name, value in values.items():
            if name in {"regions", "bounding_boxes"}:
                continue
            writer.writerow([section, name, json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else value, "", "", "", ""])
        for region in values.get("regions", []) if isinstance(values.get("regions"), list) else []:
            writer.writerow([section, "region", "", region.get("region_id"), region.get("type", "change"), region.get("area_pixels"), json.dumps(region.get("bbox_pixels") or region.get("bounding_box"))])
    response = document["authoritative_response"]
    for source in (
        ((response.get("vqa_details") or {}).get("single_image_evidence") or {}).get("regions") or [],
        (response.get("cross_modal_analysis") or {}).get("regions") or [],
    ):
        for region in source:
            writer.writerow(["regions", "region", "", region.get("region_id"), region.get("type"), region.get("area_pixels"), json.dumps(region.get("bbox_pixels") or region.get("bounding_box"))])
    grounding = response.get("grounding_result") or {}
    for index, detection in enumerate(grounding.get("detections") or [], start=1):
        value = {
            "label": detection.get("label"),
            "alignment_score": detection.get("score"),
            "bbox_normalized": detection.get("bbox_normalized"),
            "bbox_world": detection.get("bbox_world"),
            "crs": detection.get("crs"),
            "mask_url": detection.get("mask_url"),
            "source": detection.get("source"),
        }
        writer.writerow(["grounding_detections", "model_produced_detection", json.dumps(value, ensure_ascii=False), f"detection_{index}", detection.get("label"), "", json.dumps(detection.get("bbox_pixels"))])
    for index, candidate in enumerate(grounding.get("rejected_candidates") or [], start=1):
        value = {
            "label": candidate.get("label"),
            "alignment_score": candidate.get("score"),
            "box_area_ratio": candidate.get("box_area_ratio"),
            "rejection_reasons": candidate.get("rejection_reasons") or [],
            "quality": candidate.get("quality"),
            "source": candidate.get("source"),
        }
        writer.writerow(["grounding_rejected_candidates", "filtered_model_candidate", json.dumps(value, ensure_ascii=False), f"rejected_candidate_{index}", candidate.get("label"), candidate.get("quality", {}).get("box_area"), json.dumps(candidate.get("bbox_pixels"))])
    return stream.getvalue().encode("utf-8")


def _font(size: int, bold: bool = False) -> ImageFont.ImageFont:
    candidates = ["DejaVuSans-Bold.ttf" if bold else "DejaVuSans.ttf", "/System/Library/Fonts/Supplemental/Arial Bold.ttf" if bold else "/System/Library/Fonts/Supplemental/Arial.ttf"]
    for candidate in candidates:
        try:
            return ImageFont.truetype(candidate, size=size)
        except OSError:
            continue
    return ImageFont.load_default()


class _PdfPages:
    width = 1240
    height = 1754
    margin = 84
    bottom = 90

    def __init__(self, request_id: str) -> None:
        self.request_id = request_id
        self.pages: List[Image.Image] = []
        self.page = Image.new("RGB", (self.width, self.height), "white")
        self.draw = ImageDraw.Draw(self.page)
        self.y = self.margin

    def _finish(self) -> None:
        footer = f"GeoVision · SatQuery AI · {self.request_id} · Page {len(self.pages) + 1}"
        self.draw.text((self.margin, self.height - 52), footer, fill=(86, 96, 112), font=_font(18))
        self.pages.append(self.page)
        self.page = Image.new("RGB", (self.width, self.height), "white")
        self.draw = ImageDraw.Draw(self.page)
        self.y = self.margin

    def ensure(self, height: int) -> None:
        if self.y + height > self.height - self.bottom:
            self._finish()

    def rule(self) -> None:
        self.draw.line((self.margin, self.y, self.width - self.margin, self.y), fill=(208, 216, 226), width=2)
        self.y += 28

    def text(self, value: Any, size: int = 25, bold: bool = False, color: Tuple[int, int, int] = (34, 42, 55), gap: int = 14) -> None:
        text = "Unavailable" if value is None or value == "" else str(value)
        text = text.replace("\u2011", "-").replace("\u2013", "-").replace("\u2014", "-")
        font = _font(size, bold)
        max_width = self.width - self.margin * 2
        lines: List[str] = []
        for paragraph in text.splitlines() or [""]:
            words = paragraph.split() or [""]
            current = words[0]
            for word in words[1:]:
                trial = current + " " + word
                if self.draw.textlength(trial, font=font) <= max_width:
                    current = trial
                else:
                    lines.append(current)
                    current = word
            lines.append(current)
        line_height = size + 10
        self.ensure(max(line_height, len(lines) * line_height + gap))
        for line in lines:
            self.draw.text((self.margin, self.y), line, fill=color, font=font)
            self.y += line_height
        self.y += gap

    def heading(self, value: str, level: int = 1) -> None:
        self.ensure(90)
        self.text(value, size=42 if level == 1 else 31, bold=True, color=(10, 81, 122), gap=20)

    def key_value(self, key: str, value: Any) -> None:
        self.text(f"{key}: {value if value not in (None, '') else 'Unavailable'}", size=22, gap=6)

    def image(self, label: str, path: Path) -> None:
        try:
            with Image.open(path) as source:
                preview = source.convert("RGB")
                preview.thumbnail((self.width - self.margin * 2, 520), Image.Resampling.LANCZOS)
                if preview.width < 360 and preview.height < 520:
                    scale = min(360 / max(1, preview.width), 520 / max(1, preview.height))
                    preview = preview.resize((max(1, round(preview.width * scale)), max(1, round(preview.height * scale))), Image.Resampling.NEAREST)
                self.ensure(preview.height + 90)
                self.text(label, size=23, bold=True, gap=8)
                x = self.margin + ((self.width - self.margin * 2 - preview.width) // 2)
                self.page.paste(preview, (x, self.y))
                self.draw.rectangle((x, self.y, x + preview.width, self.y + preview.height), outline=(190, 201, 214), width=2)
                self.y += preview.height + 26
        except OSError:
            self.text(f"{label}: preview unavailable.", size=21)

    def output(self) -> bytes:
        self._finish()
        stream = io.BytesIO()
        self.pages[0].save(stream, format="PDF", save_all=True, append_images=self.pages[1:], resolution=150.0)
        return stream.getvalue()


def _pdf_bytes(document: Dict[str, Any]) -> bytes:
    report = document["report"]
    response = document["authoritative_response"]
    pages = _PdfPages(report["request_id"])
    pages.text("GEOVISION", 25, True, (10, 81, 122), 12)
    pages.text("SatQuery AI Mission Report", 52, True, (18, 28, 43), 24)
    pages.key_value("Analysis type", report["analysis_type"])
    pages.key_value("Request ID", report["request_id"])
    pages.key_value("Generated", report["generated_timestamp"])
    pages.key_value("Status", report["status"])
    pages.rule()
    pages.text("Evidence-backed local remote-sensing analysis with an auditable execution trace.", 27, False, (65, 75, 90), 28)

    pages.heading("Input Summary")
    for label, metadata in (("Primary image", document["input_summary"]["primary"]), ("Secondary image", document["input_summary"]["secondary"])):
        if not metadata:
            continue
        role = metadata.get("observation_role")
        pages.text(f"{role.upper() if role == 'sar' else role.title()} observation" if role else label, 26, True)
        for key in ("file_id", "observation_role", "auto_detected_modality", "effective_modality", "safe_name", "format", "width", "height", "band_count", "available_band_names", "selected_visual_bands", "band_selection_reason", "dtype", "crs", "is_georeferenced", "bounds"):
            pages.key_value(key.replace("_", " ").title(), metadata.get(key))
    compatibility = document["input_summary"].get("compatibility")
    if compatibility:
        pages.text("Compatibility", 26, True)
        for key in ("compatible", "alignment_level", "scientific_classification", "same_dimensions", "same_crs", "same_transform", "same_resolution", "same_orientation", "nodata_compatible", "overlap_ratio", "resampling_required", "recommended_action"):
            pages.key_value(key.replace("_", " ").title(), compatibility.get(key))

    pages.heading("Query and Answer")
    pages.key_value("Original query", document["user_query"]["original_query"])
    pages.key_value("Detected task", document["user_query"]["detected_task"])
    pages.key_value("Question category", document["user_query"]["question_category"])
    pages.key_value("Target concept", document["user_query"]["target_concept"])
    pages.text(document["answer"]["text"] or "No textual answer was produced.", 30, True, (21, 72, 101), 18)
    pages.key_value("Answer source", document["answer"]["source"])

    pages.heading("Evidence Products")
    pages.text(document["evidence"]["disclaimer"], 21, False, (108, 77, 16), 20)
    for item in document["evidence"]["preview_products"]:
        match = _PREVIEW_PATTERN.fullmatch(item["url"])
        path = preview_file_path(match.group(1)) if match else None
        if path:
            pages.image(item["label"], path)

    pages.heading("Statistics")
    for section, values in document["statistics"].items():
        if not values:
            continue
        pages.text(section.replace("_", " ").title(), 25, True)
        for name, value in values.items():
            if name in {"regions", "bounding_boxes"}:
                pages.key_value(name.replace("_", " ").title(), len(value) if isinstance(value, list) else value)
            elif not isinstance(value, (dict, list)):
                pages.key_value(name.replace("_", " ").title(), value)

    grounding = response.get("grounding_result") or {}
    if grounding:
        pages.heading("Grounding Detections")
        pages.key_value("Target phrase", grounding.get("target_phrase"))
        pages.key_value("Accepted detection count", len(grounding.get("detections") or []))
        pages.key_value("Rejected candidate count", len(grounding.get("rejected_candidates") or []))
        policy = grounding.get("quality_policy") or {}
        pages.key_value("Minimum operational alignment score", policy.get("minimum_alignment_score"))
        pages.key_value("Maximum localized area ratio", policy.get("maximum_localized_area_ratio"))
        pages.text(grounding.get("operational_threshold_disclaimer"), 20, False, (108, 77, 16), 14)
        pages.text("Accepted boxes are model-produced Grounding DINO localization candidates and alignment scores, not ground truth or calibrated scientific probabilities. Mask refinement is not connected.", 20, False, (108, 77, 16), 14)
        for index, detection in enumerate(grounding.get("detections") or [], start=1):
            pages.text(f"Detection {index}: {detection.get('label')} · score {detection.get('score')}", 22, True, gap=4)
            pages.key_value("Pixel box", detection.get("bbox_pixels"))
            pages.key_value("Normalized box", detection.get("bbox_normalized"))
            pages.key_value("World box", detection.get("bbox_world"))
            pages.key_value("CRS", detection.get("crs"))
        for index, candidate in enumerate(grounding.get("rejected_candidates") or [], start=1):
            pages.text(f"Rejected candidate {index}: {candidate.get('label')} · score {candidate.get('score')}", 22, True, gap=4)
            pages.key_value("Source-pixel box", candidate.get("bbox_source_xyxy"))
            pages.key_value("Box area ratio", candidate.get("box_area_ratio"))
            pages.key_value("Rejection reasons", ", ".join(candidate.get("rejection_reasons") or []))
        if grounding.get("rejected_candidates"):
            pages.text("Rejected candidates are retained for audit only and are not described as localized objects or included as accepted visual evidence.", 20, False, (108, 77, 16), 14)

    pages.heading("Confidence and Provenance")
    confidence = document.get("confidence") or {}
    pages.key_value("Confidence level", confidence.get("level"))
    pages.key_value("Calibrated score", confidence.get("score"))
    pages.key_value("Rationale", confidence.get("reason"))
    provenance = document["specialist_provenance"]
    pages.key_value("Selected tools", ", ".join(provenance.get("selected_tools") or []))
    for name, value in (provenance.get("model") or {}).items():
        pages.key_value(name.replace("_", " ").title(), value)
    if provenance.get("change_engine"):
        pages.text("Hybrid change-engine provenance", 26, True)
        for name, value in provenance["change_engine"].items():
            pages.key_value(name.replace("_", " ").title(), value)
    if provenance.get("ttp"):
        learned_change_name = provenance["ttp"].get("model") or "Learned change detector"
        for name in ("status", "model", "architecture", "training_dataset", "checkpoint", "checkpoint_fingerprint", "device", "runtime_ms", "model_load_ms", "reused_model"):
            pages.key_value(f"{learned_change_name} {name.replace('_', ' ')}".title(), provenance["ttp"].get(name))
        pages.text(f"{learned_change_name} output is a model-generated binary change prediction and is not ground truth.", 20, False, (108, 77, 16), 14)
    if provenance.get("mask_comparison"):
        pages.key_value("Mask IoU (not accuracy)", provenance["mask_comparison"].get("iou"))
        pages.key_value("Pixel agreement", provenance["mask_comparison"].get("agreement_percentage"))
        pages.key_value("Pixel disagreement", provenance["mask_comparison"].get("disagreement_percentage"))
    if provenance.get("satquery_vision_encoder"):
        sve = provenance["satquery_vision_encoder"]
        pages.text("Remote-Sensing Adaptation", 26, True)
        for name in ("model", "model_version", "backbone", "adaptation_dataset", "adapter_checksum_fingerprint", "device", "runtime_ms", "status"):
            pages.key_value(f"SVE {name.replace('_', ' ')}".title(), sve.get(name))
        for prior in sve.get("scene_priors") or []:
            pages.key_value(f"Scene prior · {prior.get('label')}", prior.get("similarity"))
        caption_consistency = sve.get("caption_consistency") or {}
        if caption_consistency:
            pages.key_value("Caption consistency", caption_consistency.get("score"))
            pages.key_value("Selected caption candidate", caption_consistency.get("selected_candidate_index"))
        vqa_consistency = sve.get("vqa_consistency") or {}
        if vqa_consistency:
            pages.key_value("VQA evidence consistency", vqa_consistency.get("state"))
        pages.text(sve.get("disclaimer"), 20, False, (108, 77, 16), 14)
    pages.key_value("Runtime", f"{provenance.get('runtime_ms')} ms")
    pages.text("Confidence values are reported only where they are real; heuristic semantic outputs are not calibrated probabilities.", 20, False, (108, 77, 16))

    pages.heading("Observable Execution Trace")
    for index, step in enumerate((document.get("execution_trace") or {}).get("steps") or [], start=1):
        pages.text(f"{index}. {step.get('tool')} · {step.get('status')} · {step.get('duration_ms')} ms", 21, False, gap=4)
        if step.get("parameters"):
            pages.text(json.dumps(step["parameters"], sort_keys=True), 17, False, (86, 96, 112), 6)

    pages.heading("Scientific Limitations")
    for limitation in document["scientific_limitations"]:
        pages.text(f"• {limitation}", 21, False, gap=8)
    # Keep this compact closing section together instead of orphaning its
    # heading at the bottom of the preceding page.
    pages.ensure(500)
    pages.heading("Reproducibility Summary")
    for key, value in document["reproducibility"].items():
        formatted = json.dumps(value, sort_keys=True) if isinstance(value, (dict, list)) else value
        pages.text(f"{key.replace('_', ' ').title()}: {formatted}", size=18, gap=4)
    return pages.output()


class ReportArtifactStore:
    def __init__(self) -> None:
        self.max_items = _configured_int("SATQUERY_REPORT_MAX_ARTIFACTS", 128, 4)
        self.ttl_seconds = _configured_int("SATQUERY_REPORT_TTL_SECONDS", DEFAULT_TTL_SECONDS, 60)
        self._items: "OrderedDict[str, Tuple[Path, float]]" = OrderedDict()
        self._lock = threading.RLock()

    def _cleanup(self) -> None:
        cutoff = time.time() - self.ttl_seconds
        for name, (path, touched) in list(self._items.items()):
            if touched < cutoff:
                path.unlink(missing_ok=True)
                self._items.pop(name, None)
        while len(self._items) > self.max_items:
            _, (path, _) = self._items.popitem(last=False)
            path.unlink(missing_ok=True)

    def put(self, name: str, data: bytes) -> Path:
        if not _ARTIFACT_PATTERN.fullmatch(name):
            raise ValueError("Unsafe report artifact filename.")
        with self._lock:
            self._cleanup()
            path = (REPORT_DIR / name).resolve()
            if path.parent != REPORT_DIR:
                raise ValueError("Unsafe report artifact path.")
            path.write_bytes(data)
            self._items[name] = (path, time.time())
            self._items.move_to_end(name)
            self._cleanup()
            return path

    def get(self, name: str) -> Optional[Path]:
        with self._lock:
            self._cleanup()
            item = self._items.get(name)
            if item is None or not item[0].is_file():
                return None
            self._items[name] = (item[0], time.time())
            self._items.move_to_end(name)
            return item[0]

    def snapshot(self) -> Dict[str, int]:
        with self._lock:
            self._cleanup()
            return {
                "stored_artifacts": len(self._items),
                "max_artifacts": self.max_items,
                "ttl_seconds": self.ttl_seconds,
            }


ARTIFACT_STORE = ReportArtifactStore()


def artifact_file_path(name: str) -> Optional[Path]:
    return ARTIFACT_STORE.get(name)


def _zip_bytes(document: Dict[str, Any], pdf: bytes, metadata: bytes, statistics: bytes) -> bytes:
    stream = io.BytesIO()
    with zipfile.ZipFile(stream, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as archive:
        archive.writestr("report/mission-report.pdf", pdf)
        archive.writestr("report/mission-report.json", metadata)
        archive.writestr("report/statistics-and-regions.csv", statistics)
        for index, item in enumerate(document["evidence"]["preview_products"], start=1):
            match = _PREVIEW_PATTERN.fullmatch(item["url"])
            path = preview_file_path(match.group(1)) if match else None
            if path:
                safe_label = re.sub(r"[^A-Za-z0-9_-]+", "_", item["label"]).strip("_")[:70]
                archive.write(path, f"evidence/{index:02d}_{safe_label}.png")
        published_references = {item["url"] for item in document["evidence"]["preview_products"]}
        evidence_manifest = {
            "schema_version": SCHEMA_VERSION,
            "request_id": document["report"]["request_id"],
            "evidence_products": [
                item for item in document["evidence"]["items"]
                if item.get("reference") in published_references
            ],
        }
        archive.writestr("evidence/manifest.json", json.dumps(evidence_manifest, indent=2, ensure_ascii=False))
        archive.writestr(
            "README.txt",
            "GeoVision SatQuery mission report package\n\n"
            f"Request ID: {document['report']['request_id']}\n"
            f"Task: {document['report']['analysis_type']}\n"
            "All evidence is model-generated or deterministic heuristic output as labelled; it is not ground truth.\n"
            "No original uploads, API keys, model-cache paths, local filesystem paths, environment values, or hidden reasoning are included.\n",
        )
    return stream.getvalue()


def generate_report(request_id: str, formats: List[ReportFormat]) -> ReportResponse:
    started = time.perf_counter()
    record = MISSION_STORE.get_by_request(request_id)
    if record is None:
        raise KeyError("The requested SatQuery result is unavailable or expired.")
    if not result_is_reportable(record.response):
        raise ValueError("This result does not contain a reportable successful, partial, or alignment outcome.")
    requested = list(dict.fromkeys(formats or [ReportFormat.PDF, ReportFormat.JSON, ReportFormat.ZIP]))
    base = f"GeoVision_SatQuery_{record.response.task.value}_{record.response.request_id}"
    names = [f"{base}.{item.value}" for item in requested]
    generated_at = utc_now()
    document = build_report_document(record.response, generated_at, names)
    metadata = json.dumps(document, indent=2, ensure_ascii=False).encode("utf-8")
    statistics = _csv_bytes(document)
    pdf = _pdf_bytes(document)
    payloads = {
        ReportFormat.PDF: pdf,
        ReportFormat.JSON: metadata,
        ReportFormat.CSV: statistics,
        ReportFormat.ZIP: _zip_bytes(document, pdf, metadata, statistics),
    }
    artifacts: List[ReportArtifact] = []
    for report_format in requested:
        name = f"{base}.{report_format.value}"
        path = ARTIFACT_STORE.put(name, payloads[report_format])
        artifacts.append(
            ReportArtifact(
                format=report_format,
                filename=name,
                url=f"/api/agent/reports/{name}",
                size_bytes=path.stat().st_size,
            )
        )
    return ReportResponse(
        request_id=request_id,
        status="success",
        schema_version=SCHEMA_VERSION,
        artifacts=artifacts,
        warnings=["Report artifacts are temporary and expire with the bounded local backend cache."],
        runtime_ms=max(0, round((time.perf_counter() - started) * 1000)),
    )