satquery_agent/comparison.py

"""Backend-authoritative, bounded mission comparison records and reports."""

from __future__ import annotations

import copy
import hashlib
import io
import json
import os
import re
import threading
import time
import uuid
import zipfile
from collections import OrderedDict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from itertools import combinations
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

from PIL import Image, ImageDraw, ImageFont

from .image_ingestion import preview_file_path, remove_preview_url, save_preview
from .models import (
    AgentResponse,
    ChangeAnalysisResponse,
    ComparabilityLevel,
    ComparabilityResult,
    ComparisonAssessmentResponse,
    ComparisonItem,
    ComparisonItemSummary,
    ComparisonLineageNode,
    ComparisonPreview,
    ComparisonReportResponse,
    ComparisonTask,
    CrossModalAnalysisResponse,
    InputMode,
    Modality,
    ReportArtifact,
    ReportFormat,
)
from .reporting import ARTIFACT_STORE, SCHEMA_VERSION
from .evidence_lifecycle import authoritative_translation_products_dict


DEFAULT_MAX_ITEMS = 32
DEFAULT_TTL_SECONDS = 30 * 60


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


def hash_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def hash_text(value: Optional[str]) -> Optional[str]:
    if not value:
        return None
    normalized = " ".join(value.lower().split())
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def composite_hash(values: Sequence[str]) -> str:
    return hashlib.sha256(":".join(values).encode("ascii")).hexdigest()


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 _safe(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", "original_name", "safe_name", "filename")):
        return None
    if isinstance(value, dict):
        output: Dict[str, Any] = {}
        for name, nested in value.items():
            cleaned = _safe(nested, name)
            if cleaned is not None:
                output[name] = cleaned
        return output
    if isinstance(value, list):
        return [_safe(nested) for nested in value]
    if isinstance(value, str):
        if value.startswith(("/Users/", "/home/", "/root/", "C:\\")):
            return "[redacted local path]"
        return re.sub(r"(?i)(api[_ -]?key|secret|password|token)\s*[:=]\s*\S+", r"\1=[redacted]", value)
    return value


@dataclass
class ComparisonIdentity:
    primary_hash: str
    secondary_hash: Optional[str] = None
    component_hashes: Tuple[str, ...] = ()
    query_hash: Optional[str] = None
    target_hash: Optional[str] = None
    dates: Tuple[Optional[str], Optional[str]] = (None, None)
    safe_parameters: Dict[str, Any] = field(default_factory=dict)
    task_family: str = ""
    output_hashes: Tuple[str, ...] = ()

    def all_input_hashes(self) -> set[str]:
        return {value for value in (self.primary_hash, self.secondary_hash, *self.component_hashes) if value}

    def public(self) -> Dict[str, Any]:
        return {
            "primary_hash_prefix": self.primary_hash[:12],
            "secondary_hash_prefix": self.secondary_hash[:12] if self.secondary_hash else None,
            "component_hash_prefixes": [value[:12] for value in self.component_hashes],
            "dates": [value for value in self.dates if value],
            "safe_parameters": _safe(self.safe_parameters),
            "identity_basis": "SHA-256 content hashes, modality, dates, safe parameters, and task family; filenames are excluded.",
        }


@dataclass
class ComparisonRecord:
    item: ComparisonItem
    identity: ComparisonIdentity
    touched_at: float
    owned_preview_urls: Tuple[str, ...] = ()


class ComparisonStore:
    """Bounded process-local store with tombstones for explicit expiry responses."""

    def __init__(self) -> None:
        self.max_items = _configured_int("SATQUERY_COMPARISON_MAX_ITEMS", DEFAULT_MAX_ITEMS, 1)
        self.ttl_seconds = _configured_int("SATQUERY_COMPARISON_TTL_SECONDS", DEFAULT_TTL_SECONDS, 60)
        self._records: "OrderedDict[str, ComparisonRecord]" = OrderedDict()
        self._expired: "OrderedDict[str, float]" = OrderedDict()
        self._lock = threading.RLock()

    def _discard(self, request_id: str, *, expired: bool = True) -> None:
        record = self._records.pop(request_id, None)
        if record is None:
            return
        for url in record.owned_preview_urls:
            remove_preview_url(url)
        if expired:
            self._expired[request_id] = time.time()
            self._expired.move_to_end(request_id)
            while len(self._expired) > self.max_items * 2:
                self._expired.popitem(last=False)

    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, item: ComparisonItem, identity: ComparisonIdentity, owned_preview_urls: Iterable[str] = ()) -> None:
        with self._lock:
            self._cleanup()
            if item.request_id in self._records:
                previous = self._records[item.request_id]
                for url in previous.owned_preview_urls:
                    if url not in owned_preview_urls:
                        remove_preview_url(url)
            item = ComparisonItem.model_validate(_safe(item.model_dump(mode="json")))
            item = item.model_copy(update={"input_identity": identity.public()})
            self._records[item.request_id] = ComparisonRecord(
                item=item.model_copy(deep=True),
                identity=copy.deepcopy(identity),
                touched_at=time.time(),
                owned_preview_urls=tuple(owned_preview_urls),
            )
            self._records.move_to_end(item.request_id)
            self._expired.pop(item.request_id, None)
            self._cleanup()

    def _lineage(self, record: ComparisonRecord) -> List[ComparisonLineageNode]:
        lineage: List[ComparisonLineageNode] = []
        for candidate in self._records.values():
            if candidate.item.request_id == record.item.request_id:
                continue
            if record.identity.primary_hash in candidate.identity.output_hashes:
                lineage.extend(candidate.item.lineage)
                lineage.append(ComparisonLineageNode(kind="derived_output", label=f"Derived from {candidate.item.display_name}", request_id=candidate.item.request_id, preview_url=(candidate.item.output_previews[0].url if candidate.item.output_previews else None)))
                break
        lineage.extend(record.item.lineage)
        seen = set()
        return [node for node in lineage if not ((node.kind, node.label, node.request_id) in seen or seen.add((node.kind, node.label, node.request_id)))]

    def get(self, request_id: str, *, touch: bool = True) -> Tuple[str, Optional[ComparisonRecord]]:
        with self._lock:
            self._cleanup()
            record = self._records.get(request_id)
            if record is None:
                return ("expired" if request_id in self._expired else "missing"), None
            if touch:
                record.touched_at = time.time()
                self._records.move_to_end(request_id)
            result = copy.deepcopy(record)
            result.item = result.item.model_copy(update={"lineage": self._lineage(record)})
            return "available", result

    def list(self) -> List[ComparisonItemSummary]:
        with self._lock:
            self._cleanup()
            output: List[ComparisonItemSummary] = []
            for record in reversed(self._records.values()):
                item = record.item
                thumbnail = (item.output_previews or item.input_previews or [None])[0]
                output.append(ComparisonItemSummary(
                    query=item.query,
                    answer_summary=item.answer,
                    request_id=item.request_id,
                    task=item.task,
                    display_name=item.display_name,
                    status=item.status,
                    created_at=item.created_at,
                    input_mode=item.input_mode,
                    modalities=item.modalities,
                    thumbnail=thumbnail,
                    execution_duration_ms=item.execution_duration_ms,
                    cached=item.cached,
                    report_available=item.report_available,
                    warning_count=len(item.warnings),
                    evidence_product_count=len(item.output_previews),
                ))
            return output

    def records(self, request_ids: Sequence[str]) -> List[ComparisonRecord]:
        records: List[ComparisonRecord] = []
        for request_id in request_ids:
            state, record = self.get(request_id)
            if record is None:
                raise LookupError(f"{state}:{request_id}")
            records.append(record)
        return records

    def mark_report(self, request_id: str) -> None:
        with self._lock:
            self._cleanup()
            record = self._records.get(request_id)
            if record:
                record.item = record.item.model_copy(update={"report_available": True})

    def clear(self) -> None:
        with self._lock:
            for request_id in list(self._records):
                self._discard(request_id, expired=False)
            self._expired.clear()

    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}


COMPARISON_STORE = ComparisonStore()


DISPLAY_NAMES = {
    ComparisonTask.CAPTIONING: "Optical Caption",
    ComparisonTask.VQA: "Controlled VQA",
    ComparisonTask.GROUNDING: "Grounding DINO",
    ComparisonTask.CHANGE: "Bi-Temporal Change Analysis",
    ComparisonTask.CHANGE_VQA: "Change VQA",
    ComparisonTask.CROSS_MODAL: "Optical–SAR Joint Analysis",
    ComparisonTask.PIX2PIX: "Pix2Pix Reconstruction",
    ComparisonTask.SARFUSIONFORMER: "SARFusionFormer Reconstruction",
    ComparisonTask.SAR_ANALYSIS: "Single-Image SAR Analysis",
}


def _task(value: str) -> ComparisonTask:
    alias = {
        "change_description": ComparisonTask.CHANGE,
        "cross_modal_analysis": ComparisonTask.CROSS_MODAL,
        "sar_water_segmentation": ComparisonTask.SAR_ANALYSIS,
        "sar_scene_analysis": ComparisonTask.SAR_ANALYSIS,
        "sar_quality_inspection": ComparisonTask.SAR_ANALYSIS,
    }.get(value)
    return alias or ComparisonTask(value)


def _preview(label: str, url: Optional[str], kind: str, metadata: Optional[Dict[str, Any]] = None, modality: Optional[Modality] = None) -> Optional[ComparisonPreview]:
    if not url:
        return None
    metadata = metadata or {}
    return ComparisonPreview(source_observation_id=metadata.get("file_id"), source_role=metadata.get("observation_role"), source_modality=metadata.get("auto_detected_modality"), label=f"{metadata['observation_role'].title()} source" if metadata.get("observation_role") and kind == "input" else label, url=url, kind=kind, width=metadata.get("width"), height=metadata.get("height"), modality=modality)


def _unique_previews(values: Iterable[Optional[ComparisonPreview]]) -> List[ComparisonPreview]:
    output: List[ComparisonPreview] = []
    seen = set()
    for value in values:
        if value is not None and value.url not in seen:
            seen.add(value.url)
            output.append(value)
    return output


def _agent_outputs(response: Dict[str, Any]) -> List[ComparisonPreview]:
    values: List[Optional[ComparisonPreview]] = []
    grounding = response.get("grounding_result") or {}
    values.append(_preview("Annotated detections", grounding.get("annotated_preview_url"), "evidence"))
    for index, detection in enumerate(grounding.get("detections") or [], start=1):
        values.append(_preview(f"Detection mask {index}", detection.get("mask_url"), "evidence"))
    change = response.get("change_analysis") or {}
    change_previews = change.get("previews") or {}
    engine_mode = ((response.get("change_engine") or change.get("change_engine") or {}).get("mode"))
    preferred_change_order = (
        ["ttp_raw_mask", "ttp_mask", "deterministic_mask", "agreement", "disagreement", "intersection", "union", "ttp_overlay", "difference"]
        if engine_mode in {"hybrid", "ttp"}
        else ["deterministic_mask", "mask", "deterministic_overlay", "overlay", "difference"]
    )
    for name in preferred_change_order:
        values.append(_preview(name.replace("_", " ").title(), change_previews.get(name), "evidence"))
    cross = response.get("cross_modal_analysis") or {}
    for product in cross.get("evidence_products") or []:
        if product.get("reference") and product.get("evidence_type") != "source":
            values.append(ComparisonPreview(label=product["label"], url=product["reference"], kind="evidence", source_observation_id=product.get("source_observation_id"), source_role=product.get("source_role"), source_modality=product.get("source_modality"), evidence_id=product.get("evidence_id")))
    for name, url in (cross.get("previews") or {}).items():
        if name not in {"optical", "sar"}:
            values.append(_preview(name.replace("_", " ").title(), url, "evidence"))
    single = ((response.get("vqa_details") or {}).get("single_image_evidence") or {})
    for name, url in (single.get("previews") or {}).items():
        values.append(_preview(name.replace("_", " ").title(), url, "evidence"))
    for product in ((response.get("sar_water_analysis") or {}).get("evidence_products") or []) + ((response.get("sar_scene_analysis") or {}).get("evidence_products") or []):
        values.append(_preview(product.get("label") or "SAR evidence", product.get("reference"), "evidence"))
    for product in authoritative_translation_products_dict(response):
        values.append(ComparisonPreview(
            label=product.get("label") or "Generated optical-like supporting evidence",
            url=product["reference"],
            kind="evidence",
            source_observation_id=product.get("source_observation_id"),
            source_role=product.get("source_role"),
            source_modality=product.get("source_modality"),
            evidence_id=product.get("evidence_id"),
        ))
    return _unique_previews(values)


def _agent_statistics(response: Dict[str, Any]) -> Dict[str, Any]:
    grounding = response.get("grounding_result") or {}
    change = response.get("change_analysis") or {}
    cross = response.get("cross_modal_analysis") or {}
    single = (((response.get("vqa_details") or {}).get("single_image_evidence")) or {})
    sar_water = response.get("sar_water_analysis") or {}
    sar_scene = response.get("sar_scene_analysis") or {}
    statistics: Dict[str, Any] = {}
    if grounding:
        statistics["grounding"] = {"detection_count": len(grounding.get("detections") or []), "top_alignment_score": (grounding.get("detections") or [{}])[0].get("score")}
    if change.get("statistics"):
        statistics["change"] = change["statistics"]
    if change.get("deterministic_statistics"):
        statistics["deterministic_change"] = change["deterministic_statistics"]
    if data_engine := (response.get("change_engine") or change.get("change_engine")):
        statistics["change_engine"] = data_engine
    if ttp := (response.get("ttp_result") or change.get("ttp_result")):
        statistics["ttp_change"] = ttp
    if mask_comparison := (response.get("mask_comparison") or change.get("mask_comparison")):
        statistics["mask_comparison"] = mask_comparison
    if semantic_summary := (response.get("semantic_change_summary") or change.get("semantic_change_summary")):
        statistics["semantic_change_summary"] = semantic_summary
    if cross.get("statistics"):
        statistics["cross_modal"] = cross["statistics"]
    if single.get("statistics"):
        statistics["single_image_evidence"] = single["statistics"]
    if (response.get("vqa_details") or {}).get("statistics_used"):
        statistics["controlled_answer"] = response["vqa_details"]["statistics_used"]
    if sar_water:
        statistics["sar_water"] = {name: sar_water.get(name) for name in ("water_detected", "image_area_percent", "candidate_pixels", "valid_pixels", "heuristic_reliability")}
    if sar_scene:
        statistics["sar_scene"] = {name: sar_scene.get(name) for name in ("valid_pixel_percent", "low_backscatter_percent", "mid_backscatter_percent", "high_backscatter_percent", "texture_index")}
    return _safe(statistics)


def _agent_limitations(response: Dict[str, Any]) -> List[str]:
    values = (
        ((response.get("caption_details") or {}).get("limitations") or [])
        + ((response.get("grounding_result") or {}).get("limitations") or [])
        + ((response.get("vqa_details") or {}).get("limitations") or [])
        + ((((response.get("cross_modal_analysis") or {}).get("method") or {}).get("limitations")) or [])
        + ((response.get("sar_water_analysis") or {}).get("limitations") or [])
        + ((response.get("sar_scene_analysis") or {}).get("limitations") or [])
        + ((response.get("ttp_result") or {}).get("limitations") or [])
    )
    return list(dict.fromkeys(values))


def record_agent_result(response: AgentResponse, *, primary_hash: str, secondary_hash: Optional[str], primary_modality: Modality, secondary_modality: Optional[Modality], primary_date: Optional[str], secondary_date: Optional[str], query: str) -> None:
    data = response.model_dump(mode="json")
    task = _task(data["task"])
    primary = data.get("primary_image_metadata") or {}
    secondary = data.get("secondary_image_metadata") or {}
    inputs = _unique_previews([
        _preview("Primary input", primary.get("preview_url"), "input", primary, primary_modality),
        _preview("Secondary input", secondary.get("preview_url"), "input", secondary, secondary_modality),
    ])
    model = data.get("model") or {}
    details = data.get("caption_details") or {}
    grounding = data.get("grounding_result") or {}
    device = details.get("device") or grounding.get("device")
    target = grounding.get("target_phrase")
    identity = ComparisonIdentity(
        primary_hash=primary_hash,
        secondary_hash=secondary_hash,
        query_hash=hash_text(query) if task in {ComparisonTask.VQA, ComparisonTask.CHANGE_VQA} else None,
        target_hash=hash_text(target) if task == ComparisonTask.GROUNDING else None,
        dates=(primary_date, secondary_date),
        safe_parameters={"input_mode": data["execution"]["input_mode"], "primary_modality": primary_modality.value, "secondary_modality": secondary_modality.value if secondary_modality else None, "date_order_supplied": bool(primary_date and secondary_date), "change_engine": (data.get("change_engine") or {}).get("mode")},
        task_family="single_image" if task in {ComparisonTask.CAPTIONING, ComparisonTask.VQA, ComparisonTask.GROUNDING, ComparisonTask.SAR_ANALYSIS} else "change" if task in {ComparisonTask.CHANGE, ComparisonTask.CHANGE_VQA} else "cross_modal",
    )
    lineage = [ComparisonLineageNode(kind="input", label="Input imagery", preview_url=inputs[0].url if inputs else None)]
    lineage.extend(ComparisonLineageNode(kind="tool", label=tool) for tool in data["execution"].get("selected_tools") or [])
    if _agent_outputs(data):
        lineage.append(ComparisonLineageNode(kind="evidence", label="Evidence products"))
    if data.get("answer"):
        lineage.append(ComparisonLineageNode(kind="answer", label="Answer"))
    item = ComparisonItem(
        query=query, request_id=data["request_id"], task=task, display_name=DISPLAY_NAMES[task], status=data["status"], created_at=(data.get("cache") or {}).get("original_generation_timestamp") or utc_now(), input_mode=InputMode(data["execution"]["input_mode"]), modalities=[value for value in (primary_modality, secondary_modality) if value], input_previews=inputs, output_previews=_agent_outputs(data), answer=data.get("answer"), statistics=_agent_statistics(data), confidence=_safe(data.get("confidence") or {}), provenance=_safe({"model": model, "method": (data.get("vqa_details") or {}).get("method"), "selection_reason": data["execution"].get("selection_reason"), "change_engine": data.get("change_engine"), "ttp": data.get("ttp_result")}), selected_tools=data["execution"].get("selected_tools") or [], execution_duration_ms=data["execution"].get("duration_ms"), device=device or (data.get("ttp_result") or {}).get("device"), warnings=_safe(data.get("warnings") or []), limitations=_agent_limitations(data), cached=bool((data.get("cache") or {}).get("cached")), input_identity=identity.public(), lineage=lineage, execution_summary=_safe({"steps": data["execution"].get("steps") or [], "validation": data["execution"].get("validation")}),
    )
    COMPARISON_STORE.put(item, identity)


def record_change_result(response: ChangeAnalysisResponse, *, primary_hash: str, secondary_hash: str, modality: Modality) -> None:
    data = response.model_dump(mode="json")
    before = data["before_metadata"]
    after = data["after_metadata"]
    engine = data.get("change_engine") or {}
    identity = ComparisonIdentity(primary_hash=primary_hash, secondary_hash=secondary_hash, dates=(data["before_date"], data["after_date"]), safe_parameters={"modality": modality.value, "date_order_supplied": True, "change_engine": engine.get("mode")}, task_family="change")
    inputs = _unique_previews([_preview("Before", data["previews"].get("before"), "input", before, modality), _preview("After", data["previews"].get("after"), "input", after, modality)])
    preview_order = ["ttp_raw_mask", "ttp_mask", "deterministic_mask", "agreement", "disagreement", "intersection", "union", "ttp_overlay", "difference"] if engine.get("mode") in {"hybrid", "ttp"} else ["deterministic_mask", "mask", "deterministic_overlay", "overlay", "difference"]
    outputs = _unique_previews(_preview(name.replace("_", " ").title(), data["previews"].get(name), "evidence", {"width": (data.get("statistics") or {}).get("analysis_width"), "height": (data.get("statistics") or {}).get("analysis_height")}) for name in preview_order)
    item = ComparisonItem(request_id=data["request_id"], task=ComparisonTask.CHANGE, display_name=DISPLAY_NAMES[ComparisonTask.CHANGE], status=data["status"], created_at=utc_now(), input_mode=InputMode.BI_TEMPORAL, modalities=[modality, modality], input_previews=inputs, output_previews=outputs, statistics=_safe({"change": data.get("statistics") or {}, "deterministic_change": data.get("deterministic_statistics") or {}, "ttp_change": data.get("ttp_result") or {}, "mask_comparison": data.get("mask_comparison") or {}}), confidence={"level": "unavailable", "score": None, "reason": ((data.get("evidence_consistency") or {}).get("disclosure") or "No calibrated change probability is produced.")}, provenance=_safe({"method": engine.get("mode") or "deterministic", "change_engine": engine, "ttp": data.get("ttp_result"), "registration": False, "reprojection": False}), selected_tools=data["execution"]["selected_tools"], execution_duration_ms=data["runtime_ms"], device=(data.get("ttp_result") or {}).get("device"), warnings=_safe(data.get("warnings") or []), limitations=["TTP and deterministic masks are model/method evidence, not ground truth.", "The workflow does not assign a semantic cause.", "Unaligned pairs are not silently registered."], input_identity=identity.public(), lineage=[ComparisonLineageNode(kind="input", label="Before and after imagery"), *[ComparisonLineageNode(kind="tool", label=tool) for tool in data["execution"]["selected_tools"]], ComparisonLineageNode(kind="evidence", label="Source-labelled learned and deterministic change evidence")], execution_summary=_safe({"steps": data["execution"].get("steps") or [], "compatibility": data.get("compatibility")}),)
    COMPARISON_STORE.put(item, identity, [preview.url for preview in inputs + outputs])


def record_cross_modal_result(response: CrossModalAnalysisResponse, *, primary_hash: str, secondary_hash: str, optical_modality: Modality, sar_modality: Modality) -> None:
    data = response.model_dump(mode="json")
    result = data["result"]
    identity = ComparisonIdentity(primary_hash=primary_hash, secondary_hash=secondary_hash, safe_parameters={"optical_modality": optical_modality.value, "sar_modality": sar_modality.value}, task_family="cross_modal")
    optical = data["optical_metadata"]
    sar = data["sar_metadata"]
    inputs = _unique_previews([_preview("Optical input", result["previews"].get("optical") or optical.get("preview_url"), "input", optical, optical_modality), _preview("SAR input", result["previews"].get("sar") or sar.get("preview_url"), "input", sar, sar_modality)])
    stats = result.get("statistics") or {}
    outputs = _unique_previews(_preview(name.replace("_", " ").title(), url, "evidence", {"width": stats.get("analysis_width"), "height": stats.get("analysis_height")}) for name, url in (result.get("previews") or {}).items() if name not in {"optical", "sar"})
    item = ComparisonItem(request_id=data["request_id"], task=ComparisonTask.CROSS_MODAL, display_name=DISPLAY_NAMES[ComparisonTask.CROSS_MODAL], status=result["status"], created_at=utc_now(), input_mode=InputMode.CROSS_MODAL, modalities=[optical_modality, sar_modality], input_previews=inputs, output_previews=outputs, statistics=_safe({"cross_modal": stats}), confidence=_safe(result.get("confidence") or {}), provenance=_safe({"method": result.get("method"), "preparation": {"optical": result.get("optical_preparation"), "sar": result.get("sar_preparation")}}), selected_tools=data["execution"]["selected_tools"], execution_duration_ms=data["execution"]["duration_ms"], warnings=_safe(result.get("warnings") or []), limitations=((result.get("method") or {}).get("limitations") or []), input_identity=identity.public(), lineage=[ComparisonLineageNode(kind="input", label="Optical and SAR pair"), ComparisonLineageNode(kind="tool", label="cross_modal_optical_sar_analyzer"), ComparisonLineageNode(kind="evidence", label="Agreement, disagreement, and joint evidence")], execution_summary=_safe({"steps": data["execution"].get("steps") or [], "compatibility": data.get("compatibility")}),)
    COMPARISON_STORE.put(item, identity, [preview.url for preview in inputs + outputs])


def store_model_preview(image: Image.Image, label: str, kind: str, modality: Optional[Modality] = None) -> ComparisonPreview:
    image = image.convert("RGB")
    return ComparisonPreview(label=label, url=save_preview(image), kind=kind, width=image.width, height=image.height, modality=modality)


def record_model_result(*, task: ComparisonTask, request_id: str, primary_hash: str, component_hashes: Sequence[str], input_previews: List[ComparisonPreview], output_previews: List[ComparisonPreview], output_hashes: Sequence[str], statistics: Dict[str, Any], execution_duration_ms: float, device: str, provenance: Dict[str, Any], warnings: Sequence[str], safe_parameters: Dict[str, Any]) -> None:
    identity = ComparisonIdentity(primary_hash=primary_hash, component_hashes=tuple(component_hashes), safe_parameters=safe_parameters, task_family="reconstruction", output_hashes=tuple(output_hashes))
    item = ComparisonItem(request_id=request_id, task=task, display_name=DISPLAY_NAMES[task], status="success", created_at=utc_now(), input_mode=InputMode.SINGLE, modalities=[Modality.SAR], input_previews=input_previews, output_previews=output_previews, statistics=_safe(statistics), confidence={"level": "unavailable", "score": None, "reason": "Reconstruction quality confidence is not calibrated; ground-truth metrics are shown only when supplied."}, provenance=_safe(provenance), selected_tools=[task.value], execution_duration_ms=max(0, round(execution_duration_ms)), device=device, warnings=_safe(list(warnings)), limitations=["Reconstruction outputs are generated products and are not ground truth.", "Runtime alone does not establish scientific accuracy."], input_identity=identity.public(), lineage=[ComparisonLineageNode(kind="input", label="SAR input", preview_url=input_previews[0].url if input_previews else None), ComparisonLineageNode(kind="tool", label=task.value), ComparisonLineageNode(kind="intermediate", label="Optical reconstruction", preview_url=output_previews[0].url if output_previews else None)], execution_summary={"measured_runtime_ms": max(0, round(execution_duration_ms))})
    COMPARISON_STORE.put(item, identity, [preview.url for preview in input_previews + output_previews])


def assess(left: ComparisonRecord, right: ComparisonRecord) -> ComparabilityResult:
    left_item, right_item = left.item, right.item
    left_hashes, right_hashes = left.identity.all_input_hashes(), right.identity.all_input_hashes()
    shared_any = bool(left_hashes & right_hashes)
    exact_pair = left.identity.primary_hash == right.identity.primary_hash and left.identity.secondary_hash == right.identity.secondary_hash
    same_family = left.identity.task_family == right.identity.task_family
    level = ComparabilityLevel.NOT_DIRECT
    reason = "The results use unrelated tasks or input identities; their task-specific measurements must not be compared as a shared score."
    warnings = ["Not directly comparable. Keep task-specific measurements in their own context."]

    reconstruction_pair = {left_item.task, right_item.task} == {ComparisonTask.PIX2PIX, ComparisonTask.SARFUSIONFORMER}
    if reconstruction_pair and shared_any:
        level, reason, warnings = ComparabilityLevel.DIRECT, "Pix2Pix and SARFusionFormer used a shared content-hash SAR identity.", ["Different reconstruction methods may produce different evidence; runtime does not identify a winner."]
    elif left_item.task == right_item.task and exact_pair:
        parameters_match = left_item.modalities == right_item.modalities and left.identity.safe_parameters == right.identity.safe_parameters
        if left_item.task in {ComparisonTask.VQA, ComparisonTask.CHANGE_VQA}:
            parameters_match = bool(left.identity.query_hash and left.identity.query_hash == right.identity.query_hash)
        if left_item.task == ComparisonTask.GROUNDING:
            parameters_match = bool(left.identity.target_hash and left.identity.target_hash == right.identity.target_hash)
        if left_item.task in {ComparisonTask.CHANGE, ComparisonTask.CHANGE_VQA}:
            parameters_match = parameters_match and left.identity.dates == right.identity.dates
        if parameters_match:
            level, reason, warnings = ComparabilityLevel.DIRECT, "The task, content-hash input identity, and required safe parameters match.", []
        else:
            level, reason, warnings = ComparabilityLevel.PARTIAL, "The imagery matches, but the question, target, dates, or other task-defining parameters differ.", ["Compare shared input evidence only; answer-specific values are not directly equivalent."]
    elif shared_any and {left_item.task, right_item.task} <= {ComparisonTask.CAPTIONING, ComparisonTask.VQA, ComparisonTask.GROUNDING, ComparisonTask.SAR_ANALYSIS}:
        level, reason, warnings = ComparabilityLevel.PARTIAL, "The same image is used by different single-image task families.", ["Caption, controlled answer, and grounding scores represent different constructs."]
    elif (left.identity.output_hashes and left.identity.output_hashes[0] in right_hashes) or (right.identity.output_hashes and right.identity.output_hashes[0] in left_hashes):
        level, reason, warnings = ComparabilityLevel.PARTIAL, "One result consumes a content-identical derived output from the other result.", ["Lineage is shared, but reconstruction and interpretation outputs remain different task types."]
    elif same_family and not shared_any:
        reason = "The task family matches, but content hashes show different input scenes or pairs."
        warnings = ["Not directly comparable because the input identity differs."]

    left_preview = (left_item.output_previews or [None])[0]
    right_preview = (right_item.output_previews or [None])[0]
    dimensions_match = bool(left_preview and right_preview and left_preview.width and left_preview.height and left_preview.width == right_preview.width and left_preview.height == right_preview.height)
    overlay = level == ComparabilityLevel.DIRECT and dimensions_match
    difference = overlay and (left_item.task == right_item.task or reconstruction_pair)
    if level == ComparabilityLevel.DIRECT and not dimensions_match and left_preview and right_preview:
        warnings.append("Overlay and difference view are disabled because output dimensions do not match or were not reported.")
    return ComparabilityResult(left_request_id=left_item.request_id, right_request_id=right_item.request_id, level=level, reason=reason, shared_inputs=shared_any, shared_task_family=same_family, warnings=warnings, overlay_allowed=overlay, difference_allowed=difference)


def assess_many(request_ids: Sequence[str]) -> ComparisonAssessmentResponse:
    records = COMPARISON_STORE.records(request_ids)
    assessments = [assess(left, right) for left, right in combinations(records, 2)]
    levels = {item.level for item in assessments}
    overall = ComparabilityLevel.NOT_DIRECT if ComparabilityLevel.NOT_DIRECT in levels else ComparabilityLevel.PARTIAL if ComparabilityLevel.PARTIAL in levels else ComparabilityLevel.DIRECT
    warnings = list(dict.fromkeys(warning for item in assessments for warning in item.warnings))
    return ComparisonAssessmentResponse(assessments=assessments, overall_level=overall, warnings=warnings)


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


def _comparison_pdf(document: Dict[str, Any]) -> bytes:
    width, height, margin = 1240, 1754, 80
    pages: List[Image.Image] = []
    page = Image.new("RGB", (width, height), "white")
    draw = ImageDraw.Draw(page)
    y = margin

    def finish() -> None:
        nonlocal page, draw, y
        draw.text((margin, height - 48), f"GeoVision · Mission Comparison · Page {len(pages) + 1}", fill=(90, 100, 114), font=_font(18))
        pages.append(page)
        page = Image.new("RGB", (width, height), "white")
        draw = ImageDraw.Draw(page)
        y = margin

    def text(value: Any, size: int = 23, bold: bool = False, color=(38, 47, 61), gap: int = 12) -> None:
        nonlocal y
        font = _font(size, bold)
        words = str(value if value not in (None, "") else "Unavailable").replace("—", "-").split()
        lines: List[str] = []
        current = ""
        for word in words:
            trial = f"{current} {word}".strip()
            if not current or draw.textlength(trial, font=font) < width - margin * 2:
                current = trial
            else:
                lines.append(current)
                current = word
        lines.append(current)
        needed = len(lines) * (size + 9) + gap
        if y + needed > height - 90:
            finish()
        for line in lines:
            draw.text((margin, y), line, fill=color, font=font)
            y += size + 9
        y += gap

    text("GEOVISION", 25, True, (10, 81, 122))
    text("Unified Mission Comparison", 48, True, (20, 29, 43), 22)
    text(f"Comparison ID: {document['comparison']['id']}")
    text(f"Generated: {document['comparison']['generated_at']}")
    text("Factual side-by-side review. No universal winner or cross-task quality score is produced.", 24, False, (99, 72, 18), 24)
    for item in document["selected_results"]:
        text(item["display_name"], 31, True, (10, 81, 122), 8)
        text(f"Task: {item['task']} · Status: {item['status']} · Runtime: {item.get('execution_duration_ms')} ms")
        text(f"Tools: {', '.join(item.get('selected_tools') or [])}", 20)
        text(f"Warnings: {len(item.get('warnings') or [])} · Evidence products: {len(item.get('output_previews') or [])}", 20)
        confidence = item.get("confidence") or {}
        text(f"Confidence basis: {confidence.get('level', 'unavailable')} · {confidence.get('reason', 'Unavailable')}", 20)
        if item.get("limitations"):
            text("Limitations: " + " | ".join(item["limitations"]), 19, False, (99, 72, 18))
    text("Comparability assessment", 32, True, (10, 81, 122), 10)
    for result in document["comparability"]:
        text(f"{result['level']}: {result['reason']}", 21)
    if document.get("user_note"):
        text("User note (user-authored)", 28, True, (10, 81, 122), 8)
        text(document["user_note"], 21)
    text("Factual conclusion", 32, True, (10, 81, 122), 10)
    for line in document["factual_conclusion"]:
        text(f"• {line}", 21)
    finish()
    stream = io.BytesIO()
    pages[0].save(stream, format="PDF", save_all=True, append_images=pages[1:], resolution=150)
    return stream.getvalue()


def generate_comparison_report(request_ids: Sequence[str], formats: Sequence[ReportFormat], user_note: Optional[str]) -> ComparisonReportResponse:
    started = time.perf_counter()
    if len(set(request_ids)) != len(request_ids):
        raise ValueError("Select distinct result IDs for a comparison report.")
    records = COMPARISON_STORE.records(request_ids)
    assessment = assess_many(request_ids)
    comparison_id = str(uuid.uuid4())
    items = [record.item.model_copy(update={"lineage": COMPARISON_STORE._lineage(record)}).model_dump(mode="json") for record in records]
    runtimes = [{"request_id": item["request_id"], "runtime_ms": item.get("execution_duration_ms")} for item in items]
    document = _safe({
        "schema_version": SCHEMA_VERSION,
        "comparison": {"id": comparison_id, "generated_at": utc_now(), "selected_count": len(items), "authoritative_backend_values": True, "winner_declared": False},
        "selected_results": items,
        "input_identity_and_lineage": [{"request_id": item["request_id"], "input_identity": item["input_identity"], "lineage": item["lineage"]} for item in items],
        "comparability": [value.model_dump(mode="json") for value in assessment.assessments],
        "runtime_and_device": [{"request_id": item["request_id"], "runtime_ms": item.get("execution_duration_ms"), "device": item.get("device")} for item in items],
        "user_note": user_note.strip() if user_note and user_note.strip() else None,
        "factual_conclusion": [f"{len(items)} stored results were selected.", f"Pairwise comparability is {assessment.overall_level.value}.", f"Measured runtimes by request ID: {json.dumps(runtimes, separators=(',', ':'))}.", "No result is ranked; task-specific measurements retain their original meaning."],
        "limitations": ["Results and artifacts are bounded, process-local, and expire.", "Runtime is a performance observation, not a scientific accuracy score."],
    })
    metadata = json.dumps(document, indent=2, ensure_ascii=False).encode("utf-8")
    pdf = _comparison_pdf(document)
    archive_stream = io.BytesIO()
    with zipfile.ZipFile(archive_stream, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as archive:
        archive.writestr("report/mission-comparison.pdf", pdf)
        archive.writestr("report/mission-comparison.json", metadata)
        for item in items:
            for index, preview in enumerate(item.get("input_previews", []) + item.get("output_previews", []), start=1):
                path = preview_file_path(preview["url"].rsplit("/", 1)[-1])
                if path:
                    label = re.sub(r"[^A-Za-z0-9_-]+", "_", preview["label"]).strip("_")[:60]
                    archive.write(path, f"evidence/{item['request_id']}/{index:02d}_{label}.png")
        archive.writestr("README.txt", "GeoVision Mission Comparison\n\nAll values are copied from authoritative stored backend results. No universal winner or cross-task quality score is generated. User notes, when present, are explicitly labelled.\n")
    payloads = {ReportFormat.PDF: pdf, ReportFormat.JSON: metadata, ReportFormat.ZIP: archive_stream.getvalue()}
    requested = list(dict.fromkeys(formats or [ReportFormat.PDF, ReportFormat.JSON, ReportFormat.ZIP]))
    if any(value not in payloads for value in requested):
        raise ValueError("Comparison reports support PDF, JSON, and ZIP formats.")
    artifacts: List[ReportArtifact] = []
    for report_format in requested:
        name = f"GeoVision_SatQuery_comparison_{comparison_id}.{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))
    for request_id in request_ids:
        COMPARISON_STORE.mark_report(request_id)
    return ComparisonReportResponse(comparison_id=comparison_id, status="success", schema_version=SCHEMA_VERSION, artifacts=artifacts, assessments=assessment.assessments, warnings=list(dict.fromkeys(assessment.warnings + ["Comparison artifacts are temporary and expire with the bounded local backend cache."])), runtime_ms=max(0, round((time.perf_counter() - started) * 1000)))