Migrate product profile reads to FastAPI

control_plane/http_app.py

@@ -1,5 +1,6 @@
 import hashlib
 import json
+import secrets
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from pathlib import Path as FilePath
@@ -30,6 +31,7 @@
 )
 from control_plane.contracts.product_environment_read_model import (
     ProductEnvironmentConfigStatus,
+    ProductReadModelStore,
 )
 from control_plane.contracts.product_profile_record import LaunchplaneProductProfileRecord
 from control_plane.contracts.preview_evidence import (
@@ -219,6 +221,23 @@ class RecentOperationsResponse(BaseModel):
     recent_previews: tuple[PreviewRecord, ...]
 
 
+class ProductProfileListResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    status: Literal["ok"] = "ok"
+    trace_id: str
+    driver_id: str
+    profiles: tuple[LaunchplaneProductProfileRecord, ...]
+
+
+class ProductProfileResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    status: Literal["ok"] = "ok"
+    trace_id: str
+    profile: LaunchplaneProductProfileRecord
+
+
 class SecretStatusBinding(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -698,6 +717,26 @@ def require_recent_operations_read_store(record_store: object) -> _RecentOperati
     return cast(_RecentOperationsReadStore, record_store)
 
 
+def require_product_profile_list_store(record_store: object) -> ProductReadModelStore:
+    list_records = getattr(record_store, "list_product_profile_records", None)
+    if not callable(list_records):
+        raise TypeError(
+            "Launchplane record store does not support product profile list reads: "
+            "list_product_profile_records"
+        )
+    return cast(ProductReadModelStore, record_store)
+
+
+def require_product_profile_read_store(record_store: object) -> ProductReadModelStore:
+    read_record = getattr(record_store, "read_product_profile_record", None)
+    if not callable(read_record):
+        raise TypeError(
+            "Launchplane record store does not support product profile reads: "
+            "read_product_profile_record"
+        )
+    return cast(ProductReadModelStore, record_store)
+
+
 def require_secret_status_read_store(record_store: object) -> control_plane_secrets.SecretReadStore:
     required_methods = (
         "read_secret_record",
@@ -981,6 +1020,34 @@ def read_identity(
             return human_identity
         raise _authentication_required_error("Authorization header is required.")
 
+    def every_code_worker_token_authorized(authorization: str) -> bool:
+        if bearer_identity_config is None:
+            return False
+        expected_token = bearer_identity_config.every_code_worker_token.strip()
+        if not expected_token:
+            return False
+        header = authorization.strip()
+        scheme, _, token = header.partition(" ")
+        bearer_token = token.strip()
+        if scheme.lower() != "bearer" or not bearer_token:
+            return False
+        return secrets.compare_digest(bearer_token, expected_token)
+
+    def read_product_profile_list_identity(
+        request: Request,
+        response: Response,
+        authorization: Annotated[str, Header(alias="Authorization")] = "",
+        cookie: Annotated[str, Header(alias="Cookie")] = "",
+    ) -> LaunchplaneIdentity | None:
+        if every_code_worker_token_authorized(authorization):
+            return None
+        return read_identity(
+            request=request,
+            response=response,
+            authorization=authorization,
+            cookie=cookie,
+        )
+
     def read_write_identity(
         authorization: Annotated[str, Header(alias="Authorization")] = "",
     ) -> LaunchplaneIdentity:
@@ -1491,6 +1558,89 @@ def read_recent_operations(
             recent_previews=recent_previews,
         )
 
+    def list_product_profiles(
+        identity: Annotated[
+            LaunchplaneIdentity | None, Depends(read_product_profile_list_identity)
+        ],
+        record_store: Annotated[object, Depends(get_record_store)],
+        driver_id: Annotated[str, Query()] = "",
+    ) -> ProductProfileListResponse:
+        trace_id = next_trace_id()
+        if identity is not None and not resolved_authz_policy_runtime.policy.allows(
+            identity=identity,
+            action="product_profile.read",
+            product="launchplane",
+            context=_LAUNCHPLANE_SERVICE_CONTEXT,
+        ):
+            raise _launchplane_http_error(
+                status_code=403,
+                trace_id=trace_id,
+                code="authorization_denied",
+                message="Workflow cannot list Launchplane product profiles.",
+            )
+        normalized_driver_id = driver_id.strip()
+        try:
+            profile_store = require_product_profile_list_store(record_store)
+            product_profile_payload = (
+                control_plane_product_read_service.build_product_profile_list_service_payload(
+                    record_store=profile_store,
+                    driver_id=normalized_driver_id,
+                )
+            )
+        except TypeError as error:
+            raise _launchplane_http_error(
+                status_code=503,
+                trace_id=trace_id,
+                code="database_storage_required",
+                message=str(error),
+            ) from error
+        payload_profiles = cast(list[object], product_profile_payload["profiles"])
+        profiles = tuple(
+            LaunchplaneProductProfileRecord.model_validate(profile) for profile in payload_profiles
+        )
+        return ProductProfileListResponse(
+            trace_id=trace_id,
+            driver_id=str(product_profile_payload["driver_id"]),
+            profiles=profiles,
+        )
+
+    def read_product_profile(
+        product: Annotated[str, Path(min_length=1, pattern=r"^\S+$")],
+        identity: Annotated[LaunchplaneIdentity, Depends(read_identity)],
+        record_store: Annotated[object, Depends(get_record_store)],
+    ) -> ProductProfileResponse:
+        trace_id = next_trace_id()
+        try:
+            profile_store = require_product_profile_read_store(record_store)
+            profile = profile_store.read_product_profile_record(product)
+        except TypeError as error:
+            raise _launchplane_http_error(
+                status_code=503,
+                trace_id=trace_id,
+                code="database_storage_required",
+                message=str(error),
+            ) from error
+        except FileNotFoundError as error:
+            raise _launchplane_http_error(
+                status_code=404,
+                trace_id=trace_id,
+                code="not_found",
+                message=str(error),
+            ) from error
+        if not resolved_authz_policy_runtime.policy.allows(
+            identity=identity,
+            action="product_profile.read",
+            product=profile.product,
+            context=_LAUNCHPLANE_SERVICE_CONTEXT,
+        ):
+            raise _launchplane_http_error(
+                status_code=403,
+                trace_id=trace_id,
+                code="authorization_denied",
+                message="Workflow cannot read the requested product profile.",
+            )
+        return ProductProfileResponse(trace_id=trace_id, profile=profile)
+
     def read_product_context_cutover_audit(
         product: Annotated[str, Path(min_length=1, pattern=r"^\S+$")],
         identity: Annotated[LaunchplaneIdentity, Depends(read_identity)],
@@ -2556,6 +2706,35 @@ def preserve_renewed_session_cookie(request: Request, response: JSONResponse) ->
         },
     )
 
+    app.add_api_route(
+        "/v1/product-profiles",
+        list_product_profiles,
+        methods=["GET"],
+        response_model=ProductProfileListResponse,
+        operation_id="list_product_profiles",
+        summary="List Launchplane product profiles",
+        responses={
+            401: {"model": LaunchplaneErrorResponse},
+            403: {"model": LaunchplaneErrorResponse},
+            503: {"model": LaunchplaneErrorResponse},
+        },
+    )
+
+    app.add_api_route(
+        "/v1/product-profiles/{product}",
+        read_product_profile,
+        methods=["GET"],
+        response_model=ProductProfileResponse,
+        operation_id="read_product_profile",
+        summary="Read a Launchplane product profile",
+        responses={
+            401: {"model": LaunchplaneErrorResponse},
+            403: {"model": LaunchplaneErrorResponse},
+            404: {"model": LaunchplaneErrorResponse},
+            503: {"model": LaunchplaneErrorResponse},
+        },
+    )
+
     app.add_api_route(
         "/v1/product-profiles/{product}/context-cutover-audit",
         read_product_context_cutover_audit,

control_plane/service.py

@@ -181,7 +181,6 @@
 from control_plane.contracts.preview_readiness_read_model import (
     build_preview_readiness_read_model,
 )
-from control_plane.contracts.product_environment_read_model import ProductReadModelStore
 from control_plane.contracts.product_profile_record import (
     LaunchplaneProductProfileRecord,
     ProductLaneProfile,
@@ -4553,10 +4552,6 @@ def _match_read_route(path: str) -> tuple[str, dict[str, str]] | None:
         return "work_graph.issue_inbox", {}
     if len(segments) == 2 and segments == ["v1", "repo-product-mapping"]:
         return "product_environment.read", {"repo_product_mapping": "true"}
-    if len(segments) == 2 and segments == ["v1", "product-profiles"]:
-        return "product_profile.read", {}
-    if len(segments) == 3 and segments[:2] == ["v1", "product-profiles"]:
-        return "product_profile.read", {"product": segments[2]}
     if len(segments) == 2 and segments == ["v1", "products"]:
         return "product_environment.read", {}
     if len(segments) == 4 and segments == [
@@ -8415,6 +8410,7 @@ def _local_admin_token_label_from_env() -> str:
 
 def _bearer_identity_config_from_env() -> BearerIdentityConfig:
     return BearerIdentityConfig(
+        every_code_worker_token=_every_code_worker_token_from_env(),
         local_admin_token=_local_admin_token_from_env(),
         local_admin_subject=_local_admin_subject_from_env(),
         local_admin_token_label=_local_admin_token_label_from_env(),
@@ -8439,8 +8435,6 @@ def _owner_agent_identity_from_bearer(environ: dict[str, object]) -> Launchplane
 
 
 def _is_every_code_worker_route(*, method: str, path: str) -> bool:
-    if method == "GET" and path == "/v1/product-profiles":
-        return True
     if method == "GET" and path == "/v1/previews/readiness":
         return True
     if method == "GET" and path == "/v1/every-code/summary":
@@ -8494,12 +8488,6 @@ def _every_code_read_payload(
     query: dict[str, list[str]],
 ) -> dict[str, object]:
     segments = [segment for segment in path.split("/") if segment]
-    if path == "/v1/product-profiles":
-        driver_id_filter = str((query.get("driver_id") or [""])[0] or "").strip()
-        return control_plane_product_read_service.build_product_profile_list_service_payload(
-            record_store=cast(ProductReadModelStore, record_store),
-            driver_id=driver_id_filter,
-        )
     every_code_store = _every_code_work_request_store(record_store)
     if path == "/v1/previews/readiness":
         repository_filter = str((query.get("repository") or [""])[0] or "").strip()
@@ -11347,78 +11335,6 @@ def start_response_with_session_cookie(status: str, headers: list[tuple[str, str
                             "targets": targets,
                         },
                     )
-                if action == "product_profile.read":
-                    if "product" in params:
-                        profile = record_store.read_product_profile_record(params["product"])
-                        if not authz_policy.allows(
-                            identity=identity,
-                            action=action,
-                            product=profile.product,
-                            context=_LAUNCHPLANE_SERVICE_CONTEXT,
-                        ):
-                            return _json_response(
-                                start_response=start_response,
-                                status_code=403,
-                                payload={
-                                    "status": "rejected",
-                                    "trace_id": request_trace_id,
-                                    "error": {
-                                        "code": "authorization_denied",
-                                        "message": "Workflow cannot read the requested product profile.",
-                                    },
-                                    "authz": _authz_diagnostic_payload(
-                                        identity=identity,
-                                        authz_policy_sha256_value=resolved_authz_policy_sha256,
-                                        authz_policy_source=resolved_authz_policy_source,
-                                    ),
-                                },
-                            )
-                        return _json_response(
-                            start_response=start_response,
-                            status_code=200,
-                            payload={
-                                "status": "ok",
-                                "trace_id": request_trace_id,
-                                "profile": profile.model_dump(mode="json"),
-                            },
-                        )
-                    if not authz_policy.allows(
-                        identity=identity,
-                        action=action,
-                        product="launchplane",
-                        context=_LAUNCHPLANE_SERVICE_CONTEXT,
-                    ):
-                        return _json_response(
-                            start_response=start_response,
-                            status_code=403,
-                            payload={
-                                "status": "rejected",
-                                "trace_id": request_trace_id,
-                                "error": {
-                                    "code": "authorization_denied",
-                                    "message": "Workflow cannot list Launchplane product profiles.",
-                                },
-                                "authz": _authz_diagnostic_payload(
-                                    identity=identity,
-                                    authz_policy_sha256_value=resolved_authz_policy_sha256,
-                                    authz_policy_source=resolved_authz_policy_source,
-                                ),
-                            },
-                        )
-                    driver_id_filter = str((query.get("driver_id") or [""])[0] or "").strip()
-                    product_profile_payload = control_plane_product_read_service.build_product_profile_list_service_payload(
-                        record_store=record_store,
-                        driver_id=driver_id_filter,
-                    )
-                    return _json_response(
-                        start_response=start_response,
-                        status_code=200,
-                        payload={
-                            "status": "ok",
-                            "trace_id": request_trace_id,
-                            **product_profile_payload,
-                        },
-                    )
                 if action == "product_environment.read":
 
                     def product_action_allowed(

control_plane/service_auth.py

@@ -94,6 +94,7 @@ def verify(self, token: str) -> GitHubActionsIdentity: ...
 class BearerIdentityConfig(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
+    every_code_worker_token: str = ""
     local_admin_token: str = ""
     local_admin_subject: str = ""
     local_admin_token_label: str = ""

docs/compatibility-retirement.md

@@ -61,10 +61,11 @@ Keep a compatibility surface only when it is one of these:
   WSGI branches are deleted; cleanup callers use the service route instead of a
   second production implementation.
 - Deployment, promotion, preview, inventory, recent-operations,
-  managed-secret status, and product context cutover audit reads use native
-  FastAPI routes for bearer-token and human-session callers. Their legacy WSGI
-  branches are deleted; direct fallback calls fail closed while the mounted
-  fallback remains for retained non-native routes.
+  managed-secret status, product-profile, and product context cutover audit
+  reads use native FastAPI routes for bearer-token and human-session callers.
+  The product-profile collection also preserves the dedicated Every Code worker
+  token. Their legacy WSGI branches are deleted; direct fallback calls fail
+  closed while the mounted fallback remains for retained non-native routes.
 - Deployment, backup-gate, promotion, preview generation, preview destroyed,
   runner-host hygiene audit, and runner-lane registration audit evidence
   ingestion use native FastAPI routes for bearer-token callers and preserve the