Narrow runtime placeholder lookup by route

control_plane/product_config.py

@@ -4,7 +4,7 @@
 import os
 from json import JSONDecodeError
 from pathlib import Path
-from typing import Literal, Protocol, cast
+from typing import Literal, Protocol, TypedDict, cast
 
 from control_plane import secrets as control_plane_secrets
 from control_plane.contracts.runtime_environment_record import RuntimeEnvironmentRecord
@@ -45,6 +45,13 @@ def list_runtime_key_safety_policy_records(
     ) -> tuple[RuntimeKeySafetyPolicyRecord, ...]: ...
 
 
+class _SecretBindingLookupKwargs(TypedDict, total=False):
+    integration: str
+    context_name: str
+    instance_name: str
+    limit: int | None
+
+
 class ProductConfigError(ValueError):
     """Operator-facing product config validation or planning failure."""
 
@@ -602,9 +609,16 @@ def _retire_disabled_runtime_secret_placeholders(
         return
     context_name = configured_binding.context.strip()
     instance_name = configured_binding.instance.strip()
+    lookup_kwargs: _SecretBindingLookupKwargs = {
+        "integration": configured_binding.integration,
+        "limit": None,
+    }
+    if context_name:
+        lookup_kwargs["context_name"] = context_name
+    if instance_name:
+        lookup_kwargs["instance_name"] = instance_name
     for binding in record_store.list_secret_bindings(
-        integration=configured_binding.integration,
-        limit=None,
+        **lookup_kwargs,
     ):
         if binding.binding_id == configured_binding.binding_id:
             continue

tests/test_runtime_environments.py

@@ -123,6 +123,7 @@ def __init__(self) -> None:
         self.secret_records: dict[str, SecretRecord] = {}
         self.secret_versions: dict[str, SecretVersion] = {}
         self.secret_bindings: dict[str, SecretBinding] = {}
+        self.secret_binding_list_calls: list[dict[str, object]] = []
         self.secret_audit_events: list[SecretAuditEvent] = []
         self.runtime_key_safety_policy_records: tuple[RuntimeKeySafetyPolicyRecord, ...] = (
             RuntimeKeySafetyPolicyRecord(
@@ -216,6 +217,14 @@ def list_secret_bindings(
         instance_name: str = "",
         limit: int | None = None,
     ) -> tuple[SecretBinding, ...]:
+        self.secret_binding_list_calls.append(
+            {
+                "integration": integration,
+                "context_name": context_name,
+                "instance_name": instance_name,
+                "limit": limit,
+            }
+        )
         bindings = tuple(
             binding
             for binding in self.secret_bindings.values()
@@ -437,6 +446,11 @@ def test_product_config_apply_retires_disabled_runtime_secret_placeholder(self)
         self.assertEqual(len(active_bindings), 1)
         self.assertEqual(active_bindings[0].binding_key, "DISCORD_TOKEN")
         self.assertEqual(active_bindings[0].status, "configured")
+        retirement_lookup = store.secret_binding_list_calls[-1]
+        self.assertEqual(retirement_lookup["integration"], "runtime_environment")
+        self.assertEqual(retirement_lookup["context_name"], "discord-blue")
+        self.assertEqual(retirement_lookup["instance_name"], "prod")
+        self.assertIsNone(retirement_lookup["limit"])
         retired_placeholder = store.secret_bindings["binding-discord-blue-placeholder"]
         self.assertEqual(retired_placeholder.integration, "retired:runtime_environment")
         self.assertEqual(retired_placeholder.binding_key, "DISCORD_TOKEN")
@@ -490,6 +504,11 @@ def test_runtime_secret_placeholder_retirement_requires_exact_route(self) -> Non
         exact_placeholder = store.secret_bindings["binding-global-placeholder"]
         self.assertEqual(exact_placeholder.integration, "retired:runtime_environment")
         self.assertEqual(exact_placeholder.status, "disabled")
+        retirement_lookup = store.secret_binding_list_calls[-1]
+        self.assertEqual(retirement_lookup["integration"], "runtime_environment")
+        self.assertEqual(retirement_lookup["context_name"], "")
+        self.assertEqual(retirement_lookup["instance_name"], "")
+        self.assertIsNone(retirement_lookup["limit"])
         unrelated_placeholder = store.secret_bindings["binding-unrelated-placeholder"]
         self.assertEqual(unrelated_placeholder.integration, "runtime_environment")
         self.assertEqual(unrelated_placeholder.status, "disabled")
@@ -542,6 +561,11 @@ def test_runtime_secret_placeholder_retirement_supports_context_route(self) -> N
         context_placeholder = store.secret_bindings["binding-context-placeholder"]
         self.assertEqual(context_placeholder.integration, "retired:runtime_environment")
         self.assertEqual(context_placeholder.status, "disabled")
+        retirement_lookup = store.secret_binding_list_calls[-1]
+        self.assertEqual(retirement_lookup["integration"], "runtime_environment")
+        self.assertEqual(retirement_lookup["context_name"], "discord-blue")
+        self.assertEqual(retirement_lookup["instance_name"], "")
+        self.assertIsNone(retirement_lookup["limit"])
         instance_placeholder = store.secret_bindings["binding-instance-placeholder"]
         self.assertEqual(instance_placeholder.integration, "runtime_environment")
         self.assertEqual(instance_placeholder.status, "disabled")