Retire satisfied onboarding secret placeholders

control_plane/workflows/product_onboarding.py

@@ -70,6 +70,13 @@ def dokploy_target_ids(self) -> tuple[DokployTargetIdRecord, ...]:
         return self.provider_target_ids
 
 
+class ProductOnboardingSecretBindingPlan(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    active_bindings: tuple[SecretBinding, ...] = ()
+    retired_bindings: tuple[SecretBinding, ...] = ()
+
+
 def build_product_profile_record(
     *, manifest: ProductOnboardingManifest, updated_at: str
 ) -> LaunchplaneProductProfileRecord:
@@ -194,23 +201,27 @@ def build_secret_bindings(
     manifest: ProductOnboardingManifest,
     updated_at: str,
     existing_bindings: tuple[SecretBinding, ...] = (),
-) -> tuple[SecretBinding, ...]:
+) -> ProductOnboardingSecretBindingPlan:
     existing_bindings_by_id = {binding.binding_id: binding for binding in existing_bindings}
-    configured_routes = {
-        (binding.integration, binding.binding_key, binding.context, binding.instance)
-        for binding in existing_bindings
-        if binding.status == "configured"
-    }
     secret_bindings: list[SecretBinding] = []
+    retired_bindings: list[SecretBinding] = []
     for binding in manifest.secret_bindings:
-        if (
-            binding.status != "configured"
-            and (binding.integration, binding.binding_key, binding.context, binding.instance)
-            in configured_routes
-        ):
-            continue
         binding_id = _secret_binding_id(product=manifest.product, binding=binding)
         existing_binding = existing_bindings_by_id.get(binding_id)
+        if binding.status != "configured" and _configured_binding_satisfies_onboarding_binding(
+            existing_bindings=existing_bindings,
+            binding=binding,
+        ):
+            if existing_binding is not None and existing_binding.status == "disabled":
+                retired_bindings.append(
+                    existing_binding.model_copy(
+                        update={
+                            "integration": f"retired:{existing_binding.integration}",
+                            "updated_at": updated_at,
+                        }
+                    )
+                )
+            continue
         secret_bindings.append(
             SecretBinding(
                 binding_id=binding_id,
@@ -224,7 +235,41 @@ def build_secret_bindings(
                 updated_at=updated_at,
             )
         )
-    return tuple(secret_bindings)
+    return ProductOnboardingSecretBindingPlan(
+        active_bindings=tuple(secret_bindings),
+        retired_bindings=tuple(retired_bindings),
+    )
+
+
+def _configured_binding_satisfies_onboarding_binding(
+    *,
+    existing_bindings: tuple[SecretBinding, ...],
+    binding: ProductOnboardingSecretBindingManifest,
+) -> bool:
+    return any(
+        existing.integration == binding.integration
+        and existing.binding_key == binding.binding_key
+        and existing.status == "configured"
+        and _binding_route_satisfies_target(
+            binding_context=existing.context,
+            binding_instance=existing.instance,
+            target_context=binding.context,
+            target_instance=binding.instance,
+        )
+        for existing in existing_bindings
+    )
+
+
+def _binding_route_satisfies_target(
+    *, binding_context: str, binding_instance: str, target_context: str, target_instance: str
+) -> bool:
+    if not binding_context:
+        return True
+    if binding_context != target_context:
+        return False
+    if not binding_instance:
+        return True
+    return binding_instance == target_instance
 
 
 def apply_product_onboarding_manifest(
@@ -246,11 +291,12 @@ def apply_product_onboarding_manifest(
     runtime_environments = build_runtime_environment_records(
         manifest=manifest, updated_at=recorded_at
     )
-    secret_bindings = build_secret_bindings(
+    secret_binding_plan = build_secret_bindings(
         manifest=manifest,
         updated_at=recorded_at,
         existing_bindings=record_store.list_secret_bindings(limit=None),
     )
+    secret_bindings = secret_binding_plan.active_bindings
     target_records_by_route = {
         (record.context, record.instance): record for record in provider_targets
     }
@@ -286,6 +332,8 @@ def apply_product_onboarding_manifest(
         record_store.write_runtime_environment_record(runtime_record)
     for binding in secret_bindings:
         record_store.write_secret_binding(binding)
+    for binding in secret_binding_plan.retired_bindings:
+        record_store.write_secret_binding(binding)
 
     return ProductOnboardingApplyResult(
         product=manifest.product,

tests/test_product_onboarding.py

@@ -1692,6 +1692,70 @@ def test_apply_product_onboarding_manifest_preserves_configured_secret_binding(
         self.assertEqual(secret_bindings[0].binding_key, "SMTP_PASSWORD")
         self.assertEqual(secret_bindings[0].status, "configured")
 
+    def test_apply_product_onboarding_manifest_retires_placeholder_when_context_binding_satisfies_instance(
+        self,
+    ) -> None:
+        with TemporaryDirectory() as temporary_directory_name:
+            store = PostgresRecordStore(
+                database_url=_sqlite_database_url(Path(temporary_directory_name) / "db.sqlite3")
+            )
+            store.ensure_schema()
+            manifest = ProductOnboardingManifest.model_validate(_manifest_payload())
+            first_result = apply_product_onboarding_manifest(
+                record_store=store,
+                manifest=manifest,
+                updated_at="2026-05-03T00:20:00Z",
+            )
+            placeholder_binding_id = first_result.secret_bindings[0].binding_id
+            store.write_secret_binding(
+                SecretBinding(
+                    binding_id="secret-runtime-environment-smtp-password-example-site-prod-binding-smtp-password",
+                    secret_id="secret-runtime-environment-smtp-password-example-site-prod",
+                    integration="runtime_environment",
+                    binding_key="SMTP_PASSWORD",
+                    context="example-site-prod",
+                    instance="",
+                    status="configured",
+                    created_at="2026-05-03T00:30:00Z",
+                    updated_at="2026-05-03T00:30:00Z",
+                )
+            )
+
+            result = apply_product_onboarding_manifest(
+                record_store=store,
+                manifest=manifest,
+                updated_at="2026-05-03T02:30:00Z",
+            )
+
+            all_secret_bindings = store.list_secret_bindings(limit=None)
+            active_secret_bindings = store.list_secret_bindings(
+                integration="runtime_environment",
+                context_name="example-site-prod",
+                instance_name="prod",
+            )
+            store.close()
+
+        retired_placeholder = next(
+            binding
+            for binding in all_secret_bindings
+            if binding.binding_id == placeholder_binding_id
+        )
+        self.assertEqual(result.secret_bindings, ())
+        self.assertEqual(retired_placeholder.integration, "retired:runtime_environment")
+        self.assertEqual(retired_placeholder.status, "disabled")
+        self.assertEqual(retired_placeholder.updated_at, "2026-05-03T02:30:00Z")
+        self.assertEqual(active_secret_bindings, ())
+        configured_bindings = [
+            binding
+            for binding in all_secret_bindings
+            if binding.integration == "runtime_environment"
+            and binding.binding_key == "SMTP_PASSWORD"
+            and binding.context == "example-site-prod"
+            and binding.instance == ""
+        ]
+        self.assertEqual(len(configured_bindings), 1)
+        self.assertEqual(configured_bindings[0].status, "configured")
+
     def test_apply_product_onboarding_manifest_blocks_conflicting_provider_target(
         self,
     ) -> None: