fix(results): gate operational AI exports on QA
GeoIntel release gates / Compile, test, contracts and builds (push) Canceled after 0s
GeoIntel release gates / Python and npm vulnerability policy (push) Canceled after 0s
GeoIntel release gates / GIS image, SBOM and container scan (push) Canceled after 0s

This commit is contained in:
Jens
2026-08-09 11:31:51 +02:00
parent b76cd1837b
commit 0209167cfd
10 changed files with 313 additions and 8 deletions
+6 -1
View File
@@ -40,7 +40,12 @@ def export_geojson(payload: GeoJsonExportRequest, db: Session = Depends(get_db))
)
if payload.export_kind == "detection_run" and payload.analysis_run_id is not None:
return envelope(
ExportService.export_detection_run_geojson(db, payload.analysis_run_id, payload.name).model_dump(mode="json")
ExportService.export_detection_run_geojson(
db,
payload.analysis_run_id,
payload.name,
intended_use=payload.intended_use,
).model_dump(mode="json")
)
if payload.export_kind == "segmentation_run" and payload.analysis_run_id is not None:
return envelope(
+4
View File
@@ -10,6 +10,7 @@ from app.schemas.operations import VectorSelectionBBox
ExportKind = Literal["dataset", "detection_run", "segmentation_run", "vector_selection"]
DetectionExportIntendedUse = Literal["review", "operational"]
MapResultMode = Literal["current", "evolution"]
@@ -21,6 +22,7 @@ class GeoJsonExportRequest(BaseModel):
name: str | None = None
bbox: VectorSelectionBBox | None = None
limit: int = 250
intended_use: DetectionExportIntendedUse = "review"
@model_validator(mode="after")
def validate_target(self) -> "GeoJsonExportRequest":
@@ -33,6 +35,8 @@ class GeoJsonExportRequest(BaseModel):
raise ValueError("bbox is required for vector selection GeoJSON exports")
if self.export_kind in {"detection_run", "segmentation_run"} and self.analysis_run_id is None:
raise ValueError("analysis_run_id is required for run GeoJSON exports")
if self.intended_use == "operational" and self.export_kind != "detection_run":
raise ValueError("operational intended_use is supported only for detection run exports")
return self
+102 -1
View File
@@ -35,6 +35,85 @@ from app.services.vector_feature_service import VectorFeatureService
class ExportService:
@staticmethod
def _detection_export_trust(db: Session, run: AnalysisRun) -> dict[str, Any]:
"""Classify persisted AI output without turning confidence into truth."""
checks = (
db.query(QualityCheck)
.filter(
QualityCheck.analysis_run_id == run.id,
QualityCheck.check_type == "detections_vs_reference",
)
.order_by(QualityCheck.created_at.desc())
.all()
)
quality_check = checks[0] if checks else None
reasons: list[str] = []
reference_dataset = None
if quality_check is None:
reasons.append("authoritative_qa_missing")
else:
findings = quality_check.findings_json if isinstance(quality_check.findings_json, dict) else {}
coverage = findings.get("coverage") if isinstance(findings.get("coverage"), dict) else {}
temporal = findings.get("temporal_compatibility") if isinstance(findings.get("temporal_compatibility"), dict) else {}
warnings = findings.get("warnings") if isinstance(findings.get("warnings"), list) else []
if quality_check.status != "ok":
reasons.append("quality_check_not_ok")
if findings.get("unsupported_geometry") is True:
reasons.append("unsupported_geometry")
false_positives = findings.get("false_positives")
false_negatives = findings.get("false_negatives")
if (
isinstance(false_positives, bool)
or not isinstance(false_positives, (int, float))
or false_positives != 0
):
reasons.append("false_positives_present")
if (
isinstance(false_negatives, bool)
or not isinstance(false_negatives, (int, float))
or false_negatives != 0
):
reasons.append("false_negatives_present")
if warnings:
reasons.append("quality_warnings_present")
if coverage.get("applied") is not True:
reasons.append("inference_coverage_not_proven")
if temporal.get("status") != "compatible":
reasons.append("temporal_compatibility_not_proven")
reference_dataset = db.get(Dataset, quality_check.reference_dataset_id)
if reference_dataset is None:
reasons.append("reference_dataset_missing")
else:
try:
DatasetConsumptionGate.assert_eligible(
reference_dataset,
purpose="reference_validation",
reference_task="building_validation",
)
except AppError:
reasons.append("reference_not_authoritative_for_buildings")
operational_use_allowed = not reasons
return {
"schema_version": "geointel.result-trust/v1",
"classification": (
"authoritative_reference_checked_ai_output"
if operational_use_allowed
else "unverified_ai_review_output"
),
"authoritative": False,
"operational_use_allowed": operational_use_allowed,
"operator_review_required": True,
"quality_check_id": str(quality_check.id) if quality_check else None,
"reference_dataset_id": str(reference_dataset.id) if reference_dataset else None,
"blocking_reasons": sorted(set(reasons)),
"limitation": (
"AI output is not ground truth. Operational use is bounded to the exact source, AOI, model and reference QA evidence."
),
}
@staticmethod
def _assert_run_source_dataset_exportable(db: Session, run: AnalysisRun) -> Dataset:
"""Block an output export when its persisted source dataset is unsafe."""
@@ -429,13 +508,33 @@ class ExportService:
return ExportService._create_response(export)
@staticmethod
def export_detection_run_geojson(db: Session, analysis_run_id: uuid.UUID, name: str | None = None) -> ExportCreateResponse:
def export_detection_run_geojson(
db: Session,
analysis_run_id: uuid.UUID,
name: str | None = None,
*,
intended_use: str = "review",
) -> ExportCreateResponse:
run = db.get(AnalysisRun, analysis_run_id)
if not run or run.analysis_type != "detection":
raise AppError(code="DETECTION_RUN_NOT_FOUND", message="Detection run not found", status_code=404)
ExportService._assert_run_source_dataset_exportable(db, run)
trust = ExportService._detection_export_trust(db, run)
if intended_use == "operational" and not trust["operational_use_allowed"]:
raise AppError(
code="DETECTION_OPERATIONAL_EXPORT_BLOCKED",
message="Operational detection export requires complete authoritative QA with no remaining errors or warnings.",
details=trust,
status_code=409,
)
feature_collection = DetectionService.detections_to_geojson(db, analysis_run_id=analysis_run_id)
feature_collection["geointel_result"] = trust
for feature in feature_collection.get("features", []):
properties = feature.get("properties") if isinstance(feature, dict) else None
if isinstance(properties, dict):
properties["result_classification"] = trust["classification"]
properties["authoritative"] = False
filename = ExportService._filename(name, f"{run.id}-detections.geojson", ".geojson")
export_path = StorageService.dataset_export_path(str(run.project_id), str(run.dataset_id or run.id), filename)
metadata = {
@@ -444,6 +543,8 @@ class ExportService:
"project_id": str(run.project_id),
"dataset_id": str(run.dataset_id) if run.dataset_id else None,
"feature_count": len(feature_collection.get("features", [])),
"intended_use": intended_use,
"result_trust": trust,
}
export = ExportService._write_json_export(
db,
@@ -4,11 +4,12 @@ import json
from datetime import datetime, timezone
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
from app.core.errors import AppError
from app.main import app
from app.models import Area, Dataset, Export, Project, QualityCheck, SourceRegistry, SourceSnapshot
from app.models import AnalysisRun, Area, Dataset, Export, Project, QualityCheck, SourceRegistry, SourceSnapshot
from app.schemas.export import ExportCreateResponse
from app.services.export_service import ExportService
from app.services.storage_service import StorageService
@@ -110,6 +111,143 @@ def _govern_fixture_dataset(dataset: Dataset) -> Dataset:
return dataset
def _authoritative_building_reference(dataset: Dataset) -> Dataset:
dataset = _govern_fixture_dataset(dataset)
dataset.dataset_role = "reference"
dataset.source_registry.usage_policy_json = {
"ground_truth_allowed": True,
"validation_authority": {"building_validation": "primary"},
}
return dataset
def test_detection_export_is_machine_labelled_as_unverified_review_output(tmp_path, monkeypatch) -> None:
project_id = uuid4()
dataset_id = uuid4()
run_id = uuid4()
source_dataset = _govern_fixture_dataset(
Dataset(
id=dataset_id,
project_id=project_id,
name="ortho.tif",
dataset_type="raster",
source="fixture",
status="ready",
)
)
run = AnalysisRun(
id=run_id,
project_id=project_id,
dataset_id=dataset_id,
analysis_type="detection",
status="success",
model_name="yolo-configured",
)
export_path = tmp_path / "detections-review.geojson"
db = FakeSession({(Dataset, dataset_id): source_dataset, (AnalysisRun, run_id): run})
monkeypatch.setattr(StorageService, "dataset_export_path", lambda *_args: str(export_path))
response = ExportService.export_detection_run_geojson(db, run_id, intended_use="review")
content = json.loads(export_path.read_text(encoding="utf-8"))
trust = content["geointel_result"]
assert response.metadata_json["intended_use"] == "review"
assert trust["classification"] == "unverified_ai_review_output"
assert trust["authoritative"] is False
assert trust["operational_use_allowed"] is False
assert trust["blocking_reasons"] == ["authoritative_qa_missing"]
def test_detection_operational_export_fails_closed_without_authoritative_qa(tmp_path, monkeypatch) -> None:
project_id = uuid4()
dataset_id = uuid4()
run_id = uuid4()
source_dataset = _govern_fixture_dataset(
Dataset(id=dataset_id, project_id=project_id, name="ortho.tif", dataset_type="raster", source="fixture", status="ready")
)
run = AnalysisRun(
id=run_id,
project_id=project_id,
dataset_id=dataset_id,
analysis_type="detection",
status="success",
model_name="yolo-configured",
)
db = FakeSession({(Dataset, dataset_id): source_dataset, (AnalysisRun, run_id): run})
monkeypatch.setattr(StorageService, "dataset_export_path", lambda *_args: str(tmp_path / "blocked.geojson"))
with pytest.raises(AppError) as exc_info:
ExportService.export_detection_run_geojson(db, run_id, intended_use="operational")
assert exc_info.value.code == "DETECTION_OPERATIONAL_EXPORT_BLOCKED"
def test_detection_operational_export_requires_zero_error_authoritative_qa(tmp_path, monkeypatch) -> None:
project_id = uuid4()
dataset_id = uuid4()
reference_id = uuid4()
run_id = uuid4()
check_id = uuid4()
source_dataset = _govern_fixture_dataset(
Dataset(id=dataset_id, project_id=project_id, name="ortho.tif", dataset_type="raster", source="fixture", status="ready")
)
reference = _authoritative_building_reference(
Dataset(
id=reference_id,
project_id=project_id,
name="grb.geojson",
dataset_type="vector",
source="fixture",
dataset_role="reference",
status="ready",
)
)
run = AnalysisRun(
id=run_id,
project_id=project_id,
dataset_id=dataset_id,
analysis_type="detection",
status="success",
model_name="yolo-configured",
)
check = QualityCheck(
id=check_id,
project_id=project_id,
analysis_run_id=run_id,
candidate_dataset_id=dataset_id,
reference_dataset_id=reference_id,
check_type="detections_vs_reference",
status="ok",
findings_json={
"false_positives": 0,
"false_negatives": 0,
"warnings": [],
"unsupported_geometry": False,
"coverage": {"applied": True},
"temporal_compatibility": {"status": "compatible"},
},
created_at=datetime.now(timezone.utc),
)
export_path = tmp_path / "detections-operational.geojson"
db = FakeSession(
{
(Dataset, dataset_id): source_dataset,
(Dataset, reference_id): reference,
(AnalysisRun, run_id): run,
(QualityCheck, check_id): check,
}
)
monkeypatch.setattr(StorageService, "dataset_export_path", lambda *_args: str(export_path))
response = ExportService.export_detection_run_geojson(db, run_id, intended_use="operational")
trust = json.loads(export_path.read_text(encoding="utf-8"))["geointel_result"]
assert response.metadata_json["intended_use"] == "operational"
assert trust["operational_use_allowed"] is True
assert trust["quality_check_id"] == str(check_id)
assert trust["reference_dataset_id"] == str(reference_id)
def test_dataset_geojson_export_persists_export_and_writes_artifact(tmp_path, monkeypatch) -> None:
project_id = uuid4()
dataset_id = uuid4()