Add digest-bound generic-web deploy recovery apply

control_plane/contracts/generic_web_deploy_recovery.py

@@ -51,6 +51,15 @@ def _validate_request(self) -> GenericWebDeployRecoveryDryRunRequest:
         return self
 
 
+class GenericWebDeployRecoveryApplyRequest(GenericWebDeployRecoveryDryRunRequest):
+    expected_recovery_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
+
+    @model_validator(mode="after")
+    def _validate_apply_request(self) -> "GenericWebDeployRecoveryApplyRequest":
+        self.expected_recovery_digest = self.expected_recovery_digest.strip()
+        return self
+
+
 class GenericWebDeployRecoveryDryRunResponse(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -76,6 +85,25 @@ class GenericWebDeployRecoveryDryRunResponse(BaseModel):
     recovery_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
 
 
+class GenericWebDeployRecoveryApplyResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: Literal[1] = 1
+    status: Literal["accepted"] = "accepted"
+    mode: Literal["apply"] = "apply"
+    trace_id: str
+    product: str
+    context: str
+    instance: str
+    reservation_state: Literal["completed", "reconcile_required"]
+    reservation_attempt: int = Field(ge=1)
+    recovery_action: GenericWebDeployRecoveryAction
+    recovery_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
+    provider_outcome: GenericWebDeployRecoveryProviderOutcome
+    provider_status: str = Field(default="", max_length=128)
+    retry_safe: bool
+
+
 def generic_web_deploy_recovery_identifier_sha256(value: str) -> str:
     return hashlib.sha256(value.strip().encode("utf-8")).hexdigest()
 

control_plane/generic_web_deploy_provider_adapter.py

@@ -119,6 +119,17 @@ def observe(
             provider_effect_phase=provider_effect_phase,
             reconciliation_key=reconciliation_key,
         )
+        return self.provider_observation_from_inspection(
+            provider_operation_key=provider_operation_key,
+            inspection=inspection,
+        )
+
+    def provider_observation_from_inspection(
+        self,
+        *,
+        provider_operation_key: str,
+        inspection: _GenericWebDeployProviderInspection,
+    ) -> ProviderObservation:
         observation = inspection.observation
         if observation.outcome != "present" or inspection.resolved_deploy_target is None:
             return ProviderObservation(