Fix Cloud Run reconciliation for resolved image indexes

CHANGELOG.md

@@ -6,6 +6,23 @@ capabilities enter through the next minor release.
 
 ## Unreleased
 
+## 0.9.0rc34 — 2026-10-01 (Control integration candidate)
+
+### Fixed
+
+- Verify Cloud Run's resolved Linux/AMD64 image against its registered immutable Artifact
+  Registry index during execution adoption and repair cleanup. The original exact comparison
+  remains the fast path; unrelated images and ambiguous index metadata still fail closed.
+- This corrects a live RC33 repair whose data work succeeded but whose terminal reconciliation
+  stopped before collecting results and restoring the job configuration.
+
+### Compatibility
+
+- The Control contract bundle and native repair protocol are unchanged. This Control candidate
+  can recover the accepted RC33 runtime execution without submitting its data work again.
+- Public PyPI remains RC20. The fixed GitHub wheel and named runtime/client artifacts form an
+  experimental integration; broader qualification and support gates remain open.
+
 ## 0.9.0rc33 — 2026-10-01 (integration candidate)
 
 ### Changed
@@ -35,6 +52,8 @@ capabilities enter through the next minor release.
 - Repair requires existing partitioned outputs with scalar business keys. Publication is atomic
   per output; earlier outputs can remain committed if a later output fails. Retained raw data
   is current data, not a historical snapshot. Live repair qualification remains outstanding.
+- Live qualification found that Control reconciliation fails when Cloud Run resolves an OCI
+  index to its platform image. Use the corrective RC34 Control candidate for that case.
 
 ## 0.9.0rc32 — 2026-08-23 (beta)
 

docs/control-profiles.md

@@ -71,6 +71,14 @@ job name, service account, image, command, and task bounds must match the deploy
 used with current Control must emit the current `runtime.completed` telemetry; the retained RC22
 worker predates that result format.
 
+When a job uses an immutable OCI index, Cloud Run reports the resolved Linux/AMD64 image on its
+execution. RC34 Control verifies that image against the exact index through Artifact Registry
+before adopting the execution or restoring repair settings. Its identity therefore needs
+`artifactregistry.dockerImages.get` on that existing repository, included in
+`roles/artifactregistry.reader`. Direct manifest references retain the exact-image fast path.
+Missing access or inconsistent index metadata stops reconciliation rather than accepting another
+image. RC33 Control lacks this index-resolution check and can leave a completed repair unresolved.
+
 With that example running, submit the graph from another terminal:
 
 ```python

docs/decisions.md

@@ -1,5 +1,16 @@
 # Engineering Decisions
 
+## 2026-10-01 — Verify Cloud Run's resolved image through Artifact Registry
+
+- Cloud Run reports a platform manifest even when the registered job uses an OCI index. Preserve
+  literal job-image matching and accept the execution's image only when authenticated Artifact
+  Registry metadata identifies it as the single Linux/AMD64 member of the exact planned index.
+- Keep this lookup inside the Cloud Run adapter and only on a same-repository image mismatch.
+  No registry framework, new dependency, cache, public DTO, or native runtime change is required.
+- Publish the Control-only correction as RC34. Preserve the failed RC33 reconciliation evidence;
+  its native data operation succeeded. Clean its cloud resources on the original deadline and
+  qualify the correction with a distinct bounded synthetic fixture.
+
 ## 2026-10-01 — Publish a fixed integration candidate before consuming its contracts
 
 - Publish RC33 as a canonical GitHub prerelease after protected merge, exact-main CI, and

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "dander-platform"
-version = "0.9.0rc33"
+version = "0.9.0rc34"
 description = "Opinionated, GCP-native, self-hosted data platform: ingest + transform + catalog behind one CLI."
 readme = "README.md"
 requires-python = ">=3.12"

src/dander/control/cloud_run_execution_backend.py

@@ -15,6 +15,7 @@
 from copy import deepcopy
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Protocol, cast
+from urllib.parse import quote
 
 from dander.control.execution_results import (
     ExecutionResultCollectionError,
@@ -504,31 +505,79 @@ def _validate_execution(
         if execution.get("name") != resource or execution.get("job") != binding.job_name:
             raise ExecutionBackendError("Cloud Run returned an unexpected execution.")
 
-    @classmethod
     def _validate_submission_execution(
-        cls,
+        self,
         plan: ExecutionPlan,
         binding: CloudRunBinding,
         execution: Mapping[str, object],
         resource: str,
         repair_payload: str | None,
     ) -> None:
-        cls._validate_execution(binding, execution, resource)
+        self._validate_execution(binding, execution, resource)
         task = execution.get("template")
         containers = task.get("containers") if isinstance(task, Mapping) else None
         container = containers[0] if isinstance(containers, list) and len(containers) == 1 else None
         if (
             not isinstance(task, Mapping)
             or not isinstance(container, Mapping)
             or task.get("serviceAccount") != binding.runtime_service_account
-            or container.get("image") != plan.image
             or container.get("args") != _execution_arguments(plan, repair_payload)
+            or not self._execution_image_matches(plan.image, container.get("image"))
         ):
             raise ExecutionBackendError(
                 "The Cloud Run execution does not match its immutable execution plan."
             )
         _require_repair_payload(task, repair_payload, _execution_arguments(plan, repair_payload))
 
+    def _execution_image_matches(self, expected: str, actual: object) -> bool:
+        if actual == expected:
+            return True
+        if (
+            not isinstance(actual, str)
+            or _ARTIFACT_IMAGE.fullmatch(actual) is None
+            or actual.rpartition("@")[0] != expected.rpartition("@")[0]
+        ):
+            return False
+        # Cloud Run resolves a configured OCI index to its Linux/AMD64 manifest.
+        # Verify that relationship against the accepted index, never a mutable tag.
+        parts = expected.split("/", maxsplit=3)
+        if len(parts) != 4:
+            return False
+        registry, project, repository, image = parts
+        region = registry.removesuffix("-docker.pkg.dev")
+        resource = (
+            f"projects/{project}/locations/{region}/repositories/{repository}/"
+            f"dockerImages/{quote(image, safe='')}"
+        )
+        try:
+            metadata = self._request_json(
+                "read image",
+                "GET",
+                f"https://artifactregistry.googleapis.com/v1/{resource}",
+                expected=(200,),
+            )
+        except _GoogleCallError as error:
+            raise ExecutionBackendError("Cloud Run image identity lookup failed.") from error
+        manifests = metadata.get("imageManifests")
+        if (
+            metadata.get("uri") != expected
+            or metadata.get("mediaType")
+            not in (
+                "application/vnd.oci.image.index.v1+json",
+                "application/vnd.docker.distribution.manifest.list.v2+json",
+            )
+            or not isinstance(manifests, list)
+        ):
+            return False
+        selected = [
+            item
+            for item in manifests
+            if isinstance(item, Mapping)
+            and item.get("os") == "linux"
+            and item.get("architecture") == "amd64"
+        ]
+        return len(selected) == 1 and selected[0].get("digest") == actual.rpartition("@")[2]
+
     def _request_json(
         self,
         operation: str,

tests/control/test_cloud_run_execution_backend.py

@@ -40,6 +40,11 @@
 JOB = "dander-hosted-graph"
 PIPELINE = "hosted_graph"
 IMAGE = f"{REGION}-docker.pkg.dev/{PROJECT}/dander/runtime@sha256:" + "b" * 64
+RESOLVED_IMAGE = IMAGE.replace("b" * 64, "d" * 64)
+IMAGE_METADATA_URL = (
+    f"https://artifactregistry.googleapis.com/v1/projects/{PROJECT}/locations/{REGION}/"
+    "repositories/dander/dockerImages/runtime%40sha256%3A" + "b" * 64
+)
 NOW = datetime(2026, 8, 26, 18, tzinfo=UTC)
 
 
@@ -92,6 +97,8 @@ class _Transport:
     executions: dict[str, dict[str, object]] = field(default_factory=dict)
     calls: list[tuple[str, str, dict[str, object]]] = field(default_factory=list)
     log_response: dict[str, object] = field(default_factory=dict)
+    image_metadata: dict[str, object] = field(default_factory=dict)
+    image_status: int = 200
     lose_patch_response: bool = False
     reject_restore: bool = False
     supersede_before_restore: bool = False
@@ -101,6 +108,8 @@ class _Transport:
 
     def request(self, method: str, url: str, **kwargs: object) -> _Response:
         self.calls.append((method, url, dict(kwargs)))
+        if method == "GET" and url == IMAGE_METADATA_URL:
+            return _Response(self.image_status, self.image_metadata)
         if url.endswith("/entries:list"):
             if self.fail_log_read:
                 raise OSError("provider transport secret")
@@ -358,6 +367,95 @@ def _execution_container(transport: _Transport, handle: BackendHandle) -> dict[s
     return containers[0]
 
 
+def _image_metadata() -> dict[str, object]:
+    return {
+        "uri": IMAGE,
+        "mediaType": "application/vnd.oci.image.index.v1+json",
+        "imageManifests": [
+            {"os": "linux", "architecture": "amd64", "digest": "sha256:" + "d" * 64},
+            {"os": "unknown", "architecture": "unknown", "digest": "sha256:" + "e" * 64},
+        ],
+    }
+
+
+def test_resolved_index_image_allows_restart_adoption_and_terminal_repair_cleanup() -> None:
+    backend, plan, transport = _backend()
+    handle = _start(backend, plan, trigger=_repair_trigger())
+    _execution_container(transport, handle)["image"] = RESOLVED_IMAGE
+    transport.image_metadata = _image_metadata()
+    restarted, _, _ = _backend(transport=transport)
+
+    assert _start(restarted, plan, trigger=_repair_trigger()) == handle
+    transport.executions[handle.execution_id].update(
+        {"completionTime": "2026-08-26T18:01:00Z", "succeededCount": 1}
+    )
+    transport.log_response = {
+        "entries": [{"timestamp": "2026-08-26T18:00:02Z", "jsonPayload": _completion_payload()}]
+    }
+
+    observed = restarted.observe(plan, handle)
+
+    assert observed.outcome is RunOutcome.SUCCEEDED
+    assert observed.results_state is ResultsState.AVAILABLE
+    assert observed.result_summary is not None
+    assert observed.cleanup_state is CleanupState.CONFIRMED
+    assert len(transport.executions) == 1
+    patches = [call[2]["json"] for call in transport.calls if call[0] == "PATCH"]
+    assert len(patches) == 2
+    restored = patches[1]
+    assert isinstance(restored, dict)
+    assert "startExecutionToken" not in restored
+    template = restored["template"]
+    assert isinstance(template, dict)
+    task = template["template"]
+    assert isinstance(task, dict)
+    containers = task["containers"]
+    assert isinstance(containers, list)
+    assert containers[0]["image"] == IMAGE
+    assert containers[0]["args"] == list(plan.execution_template.command)
+    assert containers[0]["env"] == []
+
+
+@pytest.mark.parametrize(
+    "mismatch",
+    ["index", "repository", "child", "platform", "ambiguous", "not-index", "unavailable"],
+)
+def test_resolved_image_mismatch_cannot_adopt_or_restore(mismatch: str) -> None:
+    backend, plan, transport = _backend()
+    handle = _start(backend, plan, trigger=_repair_trigger())
+    container = _execution_container(transport, handle)
+    container["image"] = RESOLVED_IMAGE
+    metadata = _image_metadata()
+    manifests = metadata["imageManifests"]
+    assert isinstance(manifests, list)
+    if mismatch == "index":
+        metadata["uri"] = RESOLVED_IMAGE
+    elif mismatch == "repository":
+        container["image"] = RESOLVED_IMAGE.replace("/dander/runtime", "/other/runtime")
+    elif mismatch == "child":
+        container["image"] = RESOLVED_IMAGE.replace("d" * 64, "f" * 64)
+    elif mismatch == "platform":
+        manifests[0]["architecture"] = "arm64"
+    elif mismatch == "ambiguous":
+        manifests.append(dict(manifests[0]))
+    elif mismatch == "not-index":
+        metadata["mediaType"] = "application/vnd.oci.image.manifest.v1+json"
+    elif mismatch == "unavailable":
+        transport.image_status = 503
+    transport.image_metadata = metadata
+    _fail_execution(transport, handle)
+
+    with pytest.raises(ExecutionBackendError):
+        _start(backend, plan, trigger=_repair_trigger())
+    with pytest.raises(ExecutionBackendError):
+        backend.observe(plan, handle)
+
+    assert len([call for call in transport.calls if call[0] == "PATCH"]) == 1
+    assert len(transport.executions) == 1
+    if mismatch == "repository":
+        assert not [call for call in transport.calls if call[1] == IMAGE_METADATA_URL]
+
+
 def test_repair_is_execution_bound_and_adopted_after_lost_response() -> None:
     backend, plan, transport = _backend(
         transport=_Transport(_job(_plan()), lose_patch_response=True)