[AZ-226] Add generated tile staging

Keep generated tiles auditable and untrusted onboard while preserving
covariance, quality, and sidecar metadata for post-flight sync.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-03 18:10:25 +03:00
parent e86084da6b
commit 2db50bc124
8 changed files with 220 additions and 2 deletions
@@ -0,0 +1,35 @@
# Batch Report
**Batch**: 5
**Tasks**: AZ-226_generated_tile_orthorectification
**Date**: 2026-05-03
## Task Results
| Task | Status | Files Modified | Tests | AC Coverage | Issues |
|------|--------|----------------|-------|-------------|--------|
| AZ-226_generated_tile_orthorectification | Done | 4 files | Pass | 3/3 ACs covered | None |
## AC Test Coverage: All covered
| AC Ref | Coverage |
|--------|----------|
| AZ-226 AC-1 | `test_eligible_frame_stages_generated_cog_and_sidecar` verifies generated COG and sidecar staging for eligible frames. |
| AZ-226 AC-2 | `test_high_covariance_generated_tile_write_is_rejected` verifies unsafe high-covariance writes are rejected and not packaged. |
| AZ-226 AC-3 | `test_sync_package_includes_manifest_delta_sidecar_covariance_and_trust_level` verifies sync package audit metadata. |
## Code Review Verdict: PASS
Review report: `_docs/03_implementation/reviews/batch_05_review.md`
## Auto-Fix Attempts: 0
## Stuck Agents: None
## Verification
- `.venv/bin/python -m black --check src tests e2e/replay` passed.
- `.venv/bin/python -m ruff check src tests e2e/replay` passed.
- `.venv/bin/python -m pytest` passed: 32 tests.
## Next Batch: AZ-228_vio_adapter, AZ-229_satellite_service_sync
@@ -0,0 +1,27 @@
# Code Review Report
**Batch**: AZ-226_generated_tile_orthorectification
**Date**: 2026-05-03
**Verdict**: PASS
## Findings
No findings.
## Spec Compliance
| Task | AC Coverage | Evidence |
|------|-------------|----------|
| AZ-226 | 3/3 covered | `tests/unit/test_tile_manager.py` verifies generated COG/sidecar staging, unsafe covariance rejection, and auditable sync package metadata. |
## Architecture Compliance
- Edits stayed inside `src/tile_manager/**` plus focused unit tests.
- Generated tile behavior consumes existing Tile Manager and shared contract patterns; no new cross-component internal imports were introduced.
- Generated outputs use `generated`/`candidate` trust levels and do not promote onboard tiles directly to trusted basemap records.
## Verification
- `.venv/bin/python -m black --check src tests e2e/replay` passed.
- `.venv/bin/python -m ruff check src tests e2e/replay` passed.
- `.venv/bin/python -m pytest` passed: 32 tests.
+1 -1
View File
@@ -9,6 +9,6 @@ tracker: jira
sub_step:
phase: 1
name: batch-loop
detail: "batch 4: AZ-223_camera_ingest_calibration, AZ-224_mavlink_gcs_gateway, AZ-225_tile_manager_cache_manifest, AZ-227_fdr_event_recorder"
detail: "batch 5: AZ-226_generated_tile_orthorectification"
retry_count: 0
cycle: 1
+8
View File
@@ -3,6 +3,10 @@
from .interfaces import LocalTileManager, TileManager
from .types import (
CacheValidationReport,
GeneratedTileCandidate,
GeneratedTileSidecar,
GeneratedTileSyncPackage,
TileGenerationRequest,
TileManifestEntry,
TileMetadataLookup,
TileValidationDecision,
@@ -11,8 +15,12 @@ from .types import (
__all__ = [
"CacheValidationReport",
"GeneratedTileCandidate",
"GeneratedTileSidecar",
"GeneratedTileSyncPackage",
"LocalTileManager",
"TileManager",
"TileGenerationRequest",
"TileManifestEntry",
"TileMetadataLookup",
"TileValidationDecision",
+53
View File
@@ -8,7 +8,11 @@ from shared.errors import ErrorEnvelope
from .types import (
CacheValidationReport,
GeneratedTileCandidate,
GeneratedTileSidecar,
GeneratedTileSyncPackage,
TileManifestEntry,
TileGenerationRequest,
TileMetadataLookup,
TileValidationDecision,
freshness_status,
@@ -40,6 +44,7 @@ class LocalTileManager:
self._trusted_by_tile_id: dict[str, CacheTileRecord] = {}
self._descriptor_by_tile_id: dict[str, str] = {}
self._tile_id_by_chunk_id: dict[str, str] = {}
self._generated_candidates: list[GeneratedTileCandidate] = []
def validate_cache(self, entries: list[TileManifestEntry]) -> CacheValidationReport:
if not self._postgis_available:
@@ -102,6 +107,54 @@ class LocalTileManager:
descriptor_ref=self._descriptor_by_tile_id[tile_id],
)
def orthorectify_frame(self, request: TileGenerationRequest) -> GeneratedTileCandidate:
if not request.frame_usable:
return GeneratedTileCandidate(accepted=False, rejection_reason="frame_not_usable")
if request.parent_covariance_m > 5.0:
return GeneratedTileCandidate(accepted=False, rejection_reason="covariance_too_high")
if request.quality_score < 0.25:
return GeneratedTileCandidate(accepted=False, rejection_reason="quality_too_low")
trust_level = "generated" if request.parent_covariance_m <= 3.0 else "candidate"
tile_id = f"generated-{request.mission_id}-{request.frame_id}"
candidate = GeneratedTileCandidate(
accepted=True,
tile_id=tile_id,
cog_ref=f"generated/{request.mission_id}/{tile_id}.cog.tif",
sidecar=GeneratedTileSidecar(
tile_id=tile_id,
parent_frame_id=request.frame_id,
parent_covariance_m=request.parent_covariance_m,
quality_score=request.quality_score,
trust_level=trust_level,
provenance=request.source_provenance,
),
)
self._generated_candidates.append(candidate)
return candidate
def package_sync(self, mission_id: str) -> GeneratedTileSyncPackage:
sidecars = tuple(
candidate.sidecar
for candidate in self._generated_candidates
if candidate.sidecar is not None
)
manifest_delta = tuple(
{
"tile_id": sidecar.tile_id,
"trust_level": sidecar.trust_level,
"parent_covariance_m": sidecar.parent_covariance_m,
"provenance": sidecar.provenance,
}
for sidecar in sidecars
)
return GeneratedTileSyncPackage(
package_ref=f"generated/{mission_id}/sync-package.json",
mission_id=mission_id,
manifest_delta=manifest_delta,
sidecars=sidecars,
)
def _validate_entry(self, entry: TileManifestEntry) -> TileValidationDecision:
if entry.signature_hash not in self._trusted_signature_hashes:
return TileValidationDecision(
+36
View File
@@ -53,6 +53,42 @@ class TileMetadataLookup(TileManagerModel):
error: ErrorEnvelope | None = None
class TileGenerationRequest(TileManagerModel):
mission_id: str = Field(min_length=1)
frame_id: str = Field(min_length=1)
image_ref: str = Field(min_length=1)
timestamp_ns: int = Field(ge=0)
parent_covariance_m: float = Field(ge=0.0)
frame_usable: bool
quality_score: float = Field(ge=0.0, le=1.0)
footprint: dict[str, float]
source_provenance: str = Field(min_length=1)
class GeneratedTileSidecar(TileManagerModel):
tile_id: str = Field(min_length=1)
parent_frame_id: str = Field(min_length=1)
parent_covariance_m: float = Field(ge=0.0)
quality_score: float = Field(ge=0.0, le=1.0)
trust_level: Literal["generated", "candidate"]
provenance: str = Field(min_length=1)
class GeneratedTileCandidate(TileManagerModel):
accepted: bool
tile_id: str | None = None
cog_ref: str | None = None
sidecar: GeneratedTileSidecar | None = None
rejection_reason: str | None = None
class GeneratedTileSyncPackage(TileManagerModel):
package_ref: str = Field(min_length=1)
mission_id: str = Field(min_length=1)
manifest_delta: tuple[dict[str, object], ...]
sidecars: tuple[GeneratedTileSidecar, ...]
def freshness_status(expires_at: datetime, now: datetime) -> Literal["fresh", "stale"]:
normalized_expiry = expires_at
if normalized_expiry.tzinfo is None:
+60 -1
View File
@@ -1,6 +1,6 @@
from datetime import datetime, timezone
from tile_manager import LocalTileManager, TileManifestEntry
from tile_manager import LocalTileManager, TileGenerationRequest, TileManifestEntry
NOW = datetime(2026, 5, 3, tzinfo=timezone.utc)
@@ -76,3 +76,62 @@ def test_tile_metadata_lookup_returns_record_or_explicit_rejection() -> None:
assert missing.found is False
assert missing.error is not None
assert missing.error.category == "validation"
def _generation_request(**overrides: object) -> TileGenerationRequest:
payload: dict[str, object] = {
"mission_id": "mission-1",
"frame_id": "frame-1",
"image_ref": "replay/frame-1.jpg",
"timestamp_ns": 10_000,
"parent_covariance_m": 2.5,
"frame_usable": True,
"quality_score": 0.8,
"footprint": {"min_lat": 49.0, "max_lat": 49.1},
"source_provenance": "nav-camera-generated",
}
payload.update(overrides)
return TileGenerationRequest.model_validate(payload)
def test_eligible_frame_stages_generated_cog_and_sidecar() -> None:
# Arrange
manager = LocalTileManager(trusted_signature_hashes={"sig:trusted"}, now=NOW)
# Act
candidate = manager.orthorectify_frame(_generation_request())
# Assert
assert candidate.accepted is True
assert candidate.cog_ref == "generated/mission-1/generated-mission-1-frame-1.cog.tif"
assert candidate.sidecar is not None
assert candidate.sidecar.trust_level == "generated"
assert candidate.sidecar.parent_covariance_m == 2.5
def test_high_covariance_generated_tile_write_is_rejected() -> None:
# Arrange
manager = LocalTileManager(trusted_signature_hashes={"sig:trusted"}, now=NOW)
# Act
candidate = manager.orthorectify_frame(_generation_request(parent_covariance_m=7.5))
# Assert
assert candidate.accepted is False
assert candidate.rejection_reason == "covariance_too_high"
assert manager.package_sync("mission-1").sidecars == ()
def test_sync_package_includes_manifest_delta_sidecar_covariance_and_trust_level() -> None:
# Arrange
manager = LocalTileManager(trusted_signature_hashes={"sig:trusted"}, now=NOW)
manager.orthorectify_frame(_generation_request())
# Act
package = manager.package_sync("mission-1")
# Assert
assert package.package_ref == "generated/mission-1/sync-package.json"
assert package.sidecars[0].parent_covariance_m == 2.5
assert package.manifest_delta[0]["trust_level"] == "generated"
assert package.manifest_delta[0]["parent_covariance_m"] == 2.5