fix(lab): сжать и адресовать RAV004 overlay

src/k1link/web/vegetation_shadow_lab_api.py

@@ -22,6 +22,7 @@ from PIL import Image
 from pydantic import BaseModel, ConfigDict, Field
 
 from k1link.laboratory.canonical_rerun_overlay import (
+    CanonicalLabOverlayArtifact,
     CanonicalLabOverlayError,
     _mask_component_boxes,
     canonical_lab_overlay,
@@ -298,12 +299,12 @@ def _build_vegetation_lab_router(
             },
         )
 
-    async def canonical_rerun_overlay_response(
+    async def canonical_rerun_overlay_artifact(
         result_id: str,
         request: CanonicalLabRerunRequest,
         *,
         expected_base_generation_sha256: str | None = None,
-    ) -> FileResponse:
+    ) -> CanonicalLabOverlayArtifact:
         """Project LAB-only evidence into the base recording's native clock."""
 
         if (
@@ -349,6 +350,11 @@ def _build_vegetation_lab_router(
                 status_code=503,
                 detail="Canonical LAB Rerun overlay failed verification",
             ) from exc
+        return artifact
+
+    def canonical_rerun_overlay_file_response(
+        artifact: CanonicalLabOverlayArtifact,
+    ) -> FileResponse:
         return FileResponse(
             artifact.path,
             media_type="application/vnd.rerun.rrd",
@@ -366,7 +372,47 @@ def _build_vegetation_lab_router(
     ) -> FileResponse:
         """Resolve the sealed sidecar for bounded non-viewer consumers."""
 
-        return await canonical_rerun_overlay_response(result_id, request)
+        artifact = await canonical_rerun_overlay_artifact(result_id, request)
+        return canonical_rerun_overlay_file_response(artifact)
+
+    @router.head("/{result_id}/canonical-overlay.rrd")
+    async def describe_canonical_rerun_overlay(
+        result_id: str,
+        application_id: Annotated[
+            Literal["nodedc_mission_core_recorded"],
+            Query(),
+        ],
+        recording_id: Annotated[
+            str,
+            Query(
+                min_length=1,
+                max_length=128,
+                pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$",
+            ),
+        ],
+        generation: Annotated[str, Query(pattern=r"^[a-f0-9]{64}$")],
+    ) -> Response:
+        """Describe the exact AI generation before Rerun opens its immutable URL."""
+
+        artifact = await canonical_rerun_overlay_artifact(
+            result_id,
+            CanonicalLabRerunRequest(
+                application_id=application_id,
+                recording_id=recording_id,
+            ),
+            expected_base_generation_sha256=generation,
+        )
+        return Response(
+            status_code=200,
+            media_type="application/vnd.rerun.rrd",
+            headers={
+                "Cache-Control": "private, no-store",
+                "Content-Length": str(artifact.byte_length),
+                "ETag": f'"{artifact.sha256}"',
+                "X-Content-Type-Options": "nosniff",
+                "X-Rerun-Format": "RRF2",
+            },
+        )
 
     @router.get("/{result_id}/canonical-overlay.rrd")
     async def stream_canonical_rerun_overlay(
@@ -384,10 +430,11 @@ def _build_vegetation_lab_router(
             ),
         ],
         generation: Annotated[str, Query(pattern=r"^[a-f0-9]{64}$")],
+        overlay_generation: Annotated[str, Query(pattern=r"^[a-f0-9]{64}$")],
     ) -> FileResponse:
         """Stream one immutable LAB sidecar through Rerun's native HTTP receiver."""
 
-        return await canonical_rerun_overlay_response(
+        artifact = await canonical_rerun_overlay_artifact(
             result_id,
             CanonicalLabRerunRequest(
                 application_id=application_id,
@@ -395,6 +442,9 @@ def _build_vegetation_lab_router(
             ),
             expected_base_generation_sha256=generation,
         )
+        if overlay_generation != artifact.sha256:
+            raise HTTPException(status_code=412, detail="Canonical overlay generation changed")
+        return canonical_rerun_overlay_file_response(artifact)
 
     @router.get("/{result_id}/timeline")
     def get_canonical_route_timeline(result_id: str) -> dict[str, object]: