fix(ota): classify deltas from prerequisite metadata

Author: supervacuus

docs/architecture.md

@@ -161,14 +161,33 @@ literal, payload-pattern, or other materialization fallback after a JSON operati
 architecture must be present, and one successful architecture never hides another architecture's materialization or
 split failure. Human stderr is retained only as bounded diagnostic data and does not drive control flow. Protocol and
 invocation violations remain exceptions; expected materialization availability is represented by typed outcomes
-instead of exceptions.
-
-Only after materialization exhausts its sources without a supported primary DSC does Symx collect `ipsw ota info/ls`
-evidence. A pure policy maps successful probe output to `delta`, `recovery`, or `unknown`. `delta` and `recovery`
-become typed extraction skip outcomes; `unknown` preserves the unavailable materialization as an extraction failure.
-If all macOS candidates are absent, this classifier runs once. The storage runner maps typed success/skip outcomes to
-persisted processing states, while exceptions are reserved for failed extraction. Symx does not own OTA cryptex mount
-lifecycle.
+instead of exceptions. `ipsw` owns the OTA cryptex mount lifecycle.
+
+### Classifying an OTA that has no usable DSC
+
+Symx classifies an OTA only after `ipsw` cannot provide a supported primary DSC. The classifier uses these sources,
+in order:
+
+1. **Request metadata:** an OTA requested for the `recovery` platform is a recovery OTA.
+2. **ZIP metadata:** Symx reads the root `Info.plist` and validates the fields it needs. A non-empty
+   `MobileAssetProperties.PrerequisiteBuild` identifies a delta OTA.
+3. **AEA metadata:** Symx asks `ipsw` to extract only `Info.plist` into a temporary directory. It accepts only small,
+   regular files that contain the expected typed metadata.
+
+`ipsw` 3.1.711 does not include `PrerequisiteBuild` in its JSON output. Some AppleArchive versions may also fail to
+reconstruct the plist from an AEA. In that case, an AEA-only compatibility fallback reads the single
+`PrereqBuild = ...` field from a successful `ipsw ota info` command. Logs record whether classification used an
+extracted plist or this fallback.
+
+The classifier follows two safety rules:
+
+- Paths such as `image_patches/` are not evidence that an OTA is a delta. Full cryptex OTAs can contain them.
+- Missing, malformed, conflicting, or unavailable metadata produces `unknown`; Symx does not guess from file names or
+  archive listings.
+
+`delta` and `recovery` are expected skip outcomes that the storage runner persists as terminal states. An `unknown`
+result preserves the original unavailable-materialization failure. For macOS, classification runs once only after all
+requested architectures are absent.
 
 A GCS extraction request owns its downloaded temporary OTA and removes it after the final macOS materialization
 attempt, before restoring split archives for symsort. `ota extract-file` does not own its input and always preserves

docs/operations.md

@@ -617,7 +617,7 @@ extract retry, provided the mirrored artifact is still present. `recovery_ota` r
 
 Existing rows are terminal skip states rather than operator emergencies. They record OTAs previously classified as referencing a DSC that the payloadv2 / Apple Archive tooling could not materialize or verify.
 
-The structured adapter does not create this state from a `payload-extract` phase plus payload/BOM inventory alone because that evidence can also accompany transient failures. Those failures now remain `symbol_extraction_failed` and visible in the default failure view until Phase 4 introduces a trusted classifier.
+The structured adapter does not create this state from a `payload-extract` phase plus payload/BOM inventory alone because that evidence can also accompany transient failures. Those failures remain `symbol_extraction_failed` and visible in the default failure view. The trusted classifier emits `delta_ota` only when it obtains a non-empty `MobileAssetProperties.PrerequisiteBuild`: directly from a ZIP root `Info.plist`, from a bounded typed `Info.plist` reconstructed from an AEA, or through the documented temporary AEA-only `PrereqBuild` text fallback required by `ipsw` 3.1.711. It does not infer delta status from `image_patches/` or other listing paths. Classification logs identify whether extracted plist metadata or the fallback was used.
 
 After a runner, macOS, `ipsw`, or AppleArchive tooling change, existing rows can be included in a curated admin extract rerun and reset to `mirrored`. They are outside the default failure view, so include `unsupported_ota_payload` in the admin state filter when reviewing them. The same explicit-filter requirement applies when retrying a misclassified `delta_ota` row.
 

symx/ota/extract.py

@@ -1,6 +1,7 @@
 """OTA extraction pipeline: DSC materialization, splitting, and symsort."""
 
 import logging
+import plistlib
 import re
 import stat
 import subprocess
@@ -27,6 +28,7 @@
 from symx.model import Arch
 from symx.fs import rmdir_if_exists
 from symx.tools import dyld_split, symsort as common_symsort
+from symx.ota.model.artifact_info import OtaArtifactInfo
 from symx.ota.model.ipsw_report import OtaDscReport, OtaDscReportFile
 from symx.ota.model.materialization import (
     OtaDscMaterializationAttempt,
@@ -63,9 +65,11 @@
     r"(?:\./)?(?:System/DriverKit/)?System/Library/(?:dyld|Caches/com\.apple\.dyld)/dyld_shared_cache_[^\s/]+"
 )
 MAX_IPSW_LISTING_PROBE_OUTPUT_CHARS = 1000
+MAX_OTA_INFO_PLIST_BYTES = 4 * 1024 * 1024
 AEA_MAGIC = b"AEA1"
 _ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m")
 _LEADING_IPSW_GLYPH_RE = re.compile(r"^\s*[•⨯]\s*")
+_PREREQUISITE_BUILD_LINE_RE = re.compile(r"^PrereqBuild\s*=\s*(\S+)\s*$", re.MULTILINE)
 IPSW_OTA_DSC_JSON_CONTRACT_RELEASE = "3.1.707"
 MACOS_OTA_DSC_ARCHITECTURES = (Arch.ARM64E, Arch.X86_64, Arch.X86_64H)
 
@@ -293,55 +297,163 @@ def symsort(dsc_split_dirs: list[Path], output_dir: Path, prefix: str, bundle_id
         raise OtaExtractError(f"Symsorter failed with {result}")
 
 
-def _collect_ota_classification_evidence(artifact: Path) -> OtaClassificationEvidence:
-    """Collect best-effort evidence only after materialization finds no usable DSC."""
-    info_result = subprocess.run(
-        ["ipsw", "ota", "info", str(artifact)],
-        capture_output=True,
-        text=True,
+def _unavailable_classification_evidence(platform: str) -> OtaClassificationEvidence:
+    return OtaClassificationEvidence(
+        platform=platform,
+        info_succeeded=False,
+        prerequisite_build=None,
+        metadata_source="unavailable",
     )
-    ls_result = subprocess.run(
-        ["ipsw", "ota", "ls", str(artifact)],
-        capture_output=True,
-        text=True,
+
+
+def _parse_ota_info_plist(data: bytes) -> OtaArtifactInfo:
+    if len(data) > MAX_OTA_INFO_PLIST_BYTES:
+        raise ValueError(f"root Info.plist is too large: {len(data)} bytes")
+    return OtaArtifactInfo.model_validate(plistlib.loads(data))
+
+
+def _read_zip_ota_classification_evidence(request: OtaExtractionRequest) -> OtaClassificationEvidence:
+    try:
+        with zipfile.ZipFile(request.local_ota) as archive:
+            info_entry = archive.getinfo("Info.plist")
+            if info_entry.file_size > MAX_OTA_INFO_PLIST_BYTES:
+                raise ValueError(f"root Info.plist is too large: {info_entry.file_size} bytes")
+            info = _parse_ota_info_plist(archive.read(info_entry))
+    except (KeyError, OSError, ValueError, plistlib.InvalidFileException, ValidationError, zipfile.BadZipFile) as error:
+        logger.warning("Could not read trusted OTA metadata from %s: %s", request.local_ota, error)
+        return _unavailable_classification_evidence(request.platform)
+
+    return OtaClassificationEvidence(
+        platform=request.platform,
+        info_succeeded=True,
+        prerequisite_build=info.prerequisite_build,
+        metadata_source="zip-info-plist",
     )
+
+
+def _extract_aea_ota_info(request: OtaExtractionRequest) -> OtaClassificationEvidence | None:
+    """Try to reconstruct a root Info.plist from an AEA without materializing symbols."""
+    with tempfile.TemporaryDirectory(suffix="_ota_info") as output_dir:
+        command = [
+            "ipsw",
+            "--no-color",
+            "ota",
+            "extract",
+            str(request.local_ota),
+            "--pattern",
+            r"^Info\.plist$",
+            "--confirm",
+            "--flat",
+            "--output",
+            output_dir,
+        ]
+        try:
+            result = subprocess.run(command, stdin=subprocess.DEVNULL, capture_output=True)
+        except OSError as error:
+            logger.warning("Could not invoke AEA OTA metadata extractor for %s: %s", request.local_ota, error)
+            return None
+
+        parsed: list[OtaArtifactInfo] = []
+        output_root = Path(output_dir).resolve()
+        for candidate in Path(output_dir).rglob("Info.plist"):
+            try:
+                resolved = candidate.resolve(strict=True)
+                if not resolved.is_relative_to(output_root):
+                    continue
+                mode = candidate.stat(follow_symlinks=False).st_mode
+                if not stat.S_ISREG(mode) or candidate.stat().st_size > MAX_OTA_INFO_PLIST_BYTES:
+                    continue
+                parsed.append(_parse_ota_info_plist(candidate.read_bytes()))
+            except (OSError, ValueError, plistlib.InvalidFileException, ValidationError):
+                continue
+
+        prerequisite_builds = {info.prerequisite_build for info in parsed}
+        if len(prerequisite_builds) == 1:
+            prerequisite_build = prerequisite_builds.pop()
+            logger.info(
+                "Read AEA OTA classification metadata from extracted Info.plist for %s (extract exit %d)",
+                request.local_ota,
+                result.returncode,
+            )
+            return OtaClassificationEvidence(
+                platform=request.platform,
+                info_succeeded=True,
+                prerequisite_build=prerequisite_build,
+                metadata_source="aea-extracted-info-plist",
+            )
+        if parsed:
+            logger.warning(
+                "AEA OTA metadata extraction returned conflicting Info.plist files for %s", request.local_ota
+            )
+        else:
+            logger.info(
+                "AEA OTA metadata extraction did not return a usable root Info.plist for %s (exit %d): %s",
+                request.local_ota,
+                result.returncode,
+                truncate_text(result.stderr) or "<empty stderr>",
+            )
+        return None
+
+
+def _read_aea_info_text_fallback(request: OtaExtractionRequest) -> OtaClassificationEvidence:
+    """Temporary fallback until structured ipsw AEA metadata includes PrerequisiteBuild."""
+    command = ["ipsw", "--no-color", "ota", "info", str(request.local_ota)]
+    try:
+        result = subprocess.run(command, capture_output=True, text=True)
+    except OSError as error:
+        logger.warning("Could not invoke AEA OTA metadata fallback for %s: %s", request.local_ota, error)
+        return _unavailable_classification_evidence(request.platform)
+    if result.returncode != 0:
+        logger.warning(
+            "Could not inspect AEA OTA metadata for %s (exit %d): %s",
+            request.local_ota,
+            result.returncode,
+            truncate_text(result.stderr) or "<empty stderr>",
+        )
+        return _unavailable_classification_evidence(request.platform)
+
+    prerequisite_builds = set(_PREREQUISITE_BUILD_LINE_RE.findall(result.stdout + "\n" + result.stderr))
+    if len(prerequisite_builds) > 1:
+        logger.warning("AEA OTA metadata fallback returned conflicting prerequisite builds for %s", request.local_ota)
+        return _unavailable_classification_evidence(request.platform)
+
+    prerequisite_build = next(iter(prerequisite_builds), None)
+    logger.info("Read AEA OTA classification metadata from ipsw info text fallback for %s", request.local_ota)
     return OtaClassificationEvidence(
-        info_returncode=info_result.returncode,
-        info_output=info_result.stdout + info_result.stderr,
-        listing_returncode=ls_result.returncode,
-        listing_output=ls_result.stdout + ls_result.stderr,
+        platform=request.platform,
+        info_succeeded=True,
+        prerequisite_build=prerequisite_build,
+        metadata_source="ipsw-info-text-fallback",
     )
 
 
+def _collect_ota_classification_evidence(request: OtaExtractionRequest) -> OtaClassificationEvidence:
+    """Read artifact metadata only after materialization finds no usable DSC."""
+    if request.platform == "recovery":
+        return OtaClassificationEvidence(
+            platform=request.platform,
+            info_succeeded=True,
+            prerequisite_build=None,
+            metadata_source="request-platform",
+        )
+    if zipfile.is_zipfile(request.local_ota):
+        return _read_zip_ota_classification_evidence(request)
+    if _ota_is_aea_archive(request.local_ota):
+        return _extract_aea_ota_info(request) or _read_aea_info_text_fallback(request)
+    return _unavailable_classification_evidence(request.platform)
+
+
 def _classify_ota_evidence(evidence: OtaClassificationEvidence) -> OtaClassification:
-    """Apply pure classification policy to collected OTA metadata and listing output."""
-    if evidence.info_returncode == 0 and (
-        "Darwin Recovery" in evidence.info_output or "RecoveryOSUpdate" in evidence.info_output
-    ):
+    """Apply pure policy to trusted request context and typed artifact metadata."""
+    if evidence.platform == "recovery":
         return OtaClassification.RECOVERY
-
-    # High-confidence delta indicators:
-    # - image_patches/: newer-style delta OTAs (e.g. iPad)
-    # - payloadv2/patches/System/Library/Caches/com.apple.dyld/: older-style deltas (e.g. Apple TV)
-    #   where the DSC itself is a binary diff
-    # Note: app_patches/ alone is not sufficient: full OTAs (e.g. watchOS, visionOS) can also
-    # contain app_patches/ alongside a full system image with a DSC.
-    if evidence.listing_returncode == 0 and (
-        "image_patches/" in evidence.listing_output
-        or "payloadv2/patches/System/Library/Caches/com.apple.dyld/" in evidence.listing_output
-    ):
+    if evidence.info_succeeded and evidence.prerequisite_build:
         return OtaClassification.DELTA
-
     return OtaClassification.UNKNOWN
 
 
-def _classify_ota(artifact: Path) -> OtaClassification:
-    try:
-        evidence = _collect_ota_classification_evidence(artifact)
-    except OSError as error:
-        logger.warning("Could not collect OTA classification evidence for %s: %s", artifact, error)
-        return OtaClassification.UNKNOWN
-    return _classify_ota_evidence(evidence)
+def _classify_ota(request: OtaExtractionRequest) -> OtaClassification:
+    return _classify_ota_evidence(_collect_ota_classification_evidence(request))
 
 
 def _parse_ota_dsc_report(
@@ -663,7 +775,7 @@ def _resolve_unavailable_materialization(
     unavailable: OtaDscUnavailable,
 ) -> OtaExtractionSkipped:
     if unavailable.exhausted_sources_without_primary:
-        classification = _classify_ota(request.local_ota)
+        classification = _classify_ota(request)
         if classification == OtaClassification.DELTA:
             return OtaExtractionSkipped(reason=OtaExtractionSkipReason.DELTA)
         if classification == OtaClassification.RECOVERY:

symx/ota/model/__init__.py

@@ -138,12 +138,12 @@ class OtaClassification(StrEnum):
 
 @dataclass(frozen=True)
 class OtaClassificationEvidence:
-    """Untrusted command results collected for pure OTA classification policy."""
+    """Trusted request context and typed artifact metadata used by classification policy."""
 
-    info_returncode: int
-    info_output: str
-    listing_returncode: int
-    listing_output: str
+    platform: str
+    info_succeeded: bool
+    prerequisite_build: str | None
+    metadata_source: str
 
 
 @dataclass(frozen=True)

symx/ota/model/artifact_info.py

@@ -0,0 +1,23 @@
+"""Typed subset of the root Info.plist embedded in ZIP OTA artifacts."""
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class OtaMobileAssetProperties(BaseModel):
+    """Mobile asset fields used to identify prerequisite/delta OTAs."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    prerequisite_build: str = Field(default="", alias="PrerequisiteBuild")
+
+
+class OtaArtifactInfo(BaseModel):
+    """Trusted classification fields from an OTA's root Info.plist."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    mobile_asset_properties: OtaMobileAssetProperties = Field(alias="MobileAssetProperties")
+
+    @property
+    def prerequisite_build(self) -> str | None:
+        return self.mobile_asset_properties.prerequisite_build or None