# PP-1388 — Feature Controls Hydration (Option 2: Policy-driven)

**Spike reference**: [SPIKE: How do we want to hydrate Feature Control data for an account or subaccount?](https://www.notion.so/SPIKE-PP-1388-How-do-we-want-to-hydrate-Feature-Control-data-for-an-account-or-subaccount-31097177520f80989e60d7d67a4c1513)

> **Note**: Workstation roles hydration is intentionally excluded from this epic and deferred to PP-1411.

---

## Open Questions

1. **Scope of implementation** — Should this apply to all `check_resources` endpoints or just `check_my_resources`?

---

## Tickets

### Ticket Summary

| Ticket     | Title                                                                                    | Service     | Blocked by  |
|------------|------------------------------------------------------------------------------------------|-------------|-------------|
| Ticket 01a | `account/models/feature.py` — SQL query functions                                        | ows-account | —           |
| Ticket 01b | `account/logic/feature.py` — logic wrappers                                              | ows-account | Ticket 01a (PP-1426) |
| Ticket 01c | Handler + schema layer for vendor/subaccount feature endpoints                           | ows-account | Ticket 01b (PP-1427) |
| Ticket 02  | `CerbosPolicyParser` + `PolicyMetadataDatabase` + cache key constant                     | ows-pdp     | —           |
| Ticket 03  | `seed_policy_metadata_cache` CLI command + `make` target                                 | ows-pdp     | Ticket 02 (PP-1429) |
| Ticket 04  | Jenkinsfile `Seed QA/Prod Policy Metadata Cache` stages                                  | ows-pdp     | Ticket 03 (PP-1430) |
| Ticket 05  | `datasources.py` reads `PolicyMetadataDatabase` from Redis at startup                    | ows-pdp     | Ticket 02 (PP-1429), Ticket 03 (PP-1430), Ticket 10 (PP-1438) |
| Ticket 06  | `Resource.needs_account_feature_controls_lookup()`                                       | ows-pdp     | Ticket 02 (PP-1429) |
| Ticket 07  | `OwsAccountClient` response models + client methods                                      | ows-pdp     | Ticket 01c (PP-1428) |
| Ticket 08  | `VendorFeatureControlsProxy`                                                             | ows-pdp     | Ticket 07 (PP-1434) |
| Ticket 09a | `_hydrate_resources_with_feature_controls_as_needed` + `check_resources` signature       | ows-pdp     | Ticket 05 (PP-1432), Ticket 06 (PP-1433), Ticket 08 (PP-1435), Ticket 10 (PP-1438) |
| Ticket 09b | Endpoint injection of `policy_metadata_db` dependency                                    | ows-pdp     | Ticket 09a (PP-1436) |
| Ticket 10  | Feature flag `pp_vendor_features_lookup` constant + Split.io flag creation               | ows-pdp     | —           |
| Ticket 11  | `POST /infra/tenant/feature-controls/` endpoint + `TenantFeatureControls` schema         | ows-pdp     | Ticket 08 (PP-1435) |
| Ticket 12  | Redis cache + invalidation for vendor features lookup endpoints                          | ows-account | Ticket 01c (PP-1428) |

### Dependency Graph

```
01a (ows-account: model layer)
 └── 01b (ows-account: logic layer)
      └── 01c (ows-account: handler + schema layer)
           ├── 07 (OwsAccountClient models + methods)
           └── 12 (Redis cache + invalidation)
                └── 08 (VendorFeatureControlsProxy)
                     └── 09a (hydration fn + check_resources) ◄─┐
                          └── 09b (endpoint injection)           │
                                                                 │
02 (PolicyMetadataDatabase + parser)                             │
 ├── 03 (CLI seed + Makefile)                                    │
 │    ├── 04 (Jenkinsfile stages)                                │
 │    └── 05 (startup Redis read) ───────────────────────────────┤
 └── 06 (Resource class method) ────────────────────────────────┘

10 (Feature flag pp_vendor_features_lookup) — independent; blocks 05 and 09a

08 (VendorFeatureControlsProxy)
 └── 11 (infra /infra/tenant/feature-controls/ endpoint)
```

- Tickets **01a**, **02**, and **10** are independent starting points.
  - Ticket **09b** is the final integration step — it unblocks only after 09a is complete (which itself requires 05, 06, 08, and 10).
  - Ticket **11** is independently shippable once Ticket 08 is merged.
  - Ticket **12** is independently shippable once Ticket 01c is merged; it does not block any other ticket.

---

### Ticket 01a — ows-account: Model Layer — SQL Query Functions for Feature Controls

**Service**: ows-account | **Blocked by**: — | **Blocks**: Ticket 01b (PP-1427)

#### Summary

Add two SQL query functions to `account/models/feature.py` that return enabled feature IDs for
vendor and subaccount tenants by UUID. These are pure data-layer functions with no HTTP concerns.
The vendor query joins `vendor → features` and excludes rows in `vendor_restricted_features`. The
subaccount query does the same via `subaccount → vendor_restricted_features`, resolving to the
parent vendor internally.

#### Acceptance Criteria

- `get_enabled_feature_ids_for_vendor_uuids(vendor_uuids, session)` returns `dict[str, list[int]]` keyed by vendor UUID.
- `get_enabled_feature_ids_for_subaccount_uuids(subaccount_uuids, session)` returns `dict[str, list[int]]` keyed by subaccount UUID.
  - Both functions return only UUIDs present in their respective tables (missing UUIDs omitted).
  - An empty list for a UUID means the vendor/parent vendor has no restricted features (all features enabled).
  - Unit tests cover: known UUID with features, known UUID with no restricted features (empty list), unknown UUID omitted from result.

#### Implementation Details

**`account/models/feature.py`** — new SQL query functions

```python
def get_enabled_feature_ids_for_vendor_uuids(vendor_uuids, session):
    """
    SELECT v.vendor_uuid, f.feature_id
    FROM vendor v
    CROSS JOIN features f
    WHERE v.vendor_uuid IN ({placeholders})
    AND NOT EXISTS (
        SELECT 1
        FROM vendor_restricted_features vrf
        WHERE vrf.vendor_id = v.vendor_id
        AND vrf.feature_id = f.feature_id
    )
    ORDER BY v.vendor_uuid, f.feature_id
    """
    # Returns dict[str, list[int]]: vendor_uuid -> [feature_id, ...]

def get_enabled_feature_ids_for_subaccount_uuids(subaccount_uuids, session):
    """
    SELECT sa.subaccount_uuid, f.feature_id
    FROM subaccount sa
    CROSS JOIN features f
    WHERE sa.subaccount_uuid IN ({placeholders})
    AND NOT EXISTS (
        SELECT 1
        FROM vendor_restricted_features vrf
        WHERE vrf.vendor_id = sa.vendor_id
        AND vrf.feature_id = f.feature_id
    )
    ORDER BY sa.subaccount_uuid, f.feature_id
    """
    # Returns dict[str, list[int]]: subaccount_uuid -> [feature_id, ...]
```

---

### Ticket 01b — ows-account: Logic Layer — Feature Control Lookup Wrappers

**Service**: ows-account | **Blocked by**: Ticket 01a (PP-1426) | **Blocks**: Ticket 01c (PP-1428)

#### Summary

Add two logic-layer wrapper functions to `account/logic/feature.py` that call the model functions
from Ticket 01a and return standard `response.Response` objects. Follows the existing pattern for
other lookup logic functions in this module.

#### Acceptance Criteria

- `lookup_features_by_vendor_uuids(uuids)` calls the model and returns `response.Response({"vendors": [...]})` as an ordered list matching the input UUID order.
- `lookup_features_by_subaccount_uuids(uuids)` calls the model and returns `response.Response({"subaccounts": [...]})` as an ordered list matching the input UUID order.
- UUIDs not found in the database are included in the list with `feature_ids: []`.
- Unit tests cover the happy path for each function.

#### Implementation Details

**`account/logic/feature.py`** — logic wrappers

```python
def lookup_features_by_vendor_uuids(uuids):
    # Calls model, maps results back to input UUID order,
    # defaulting to feature_ids=[] for unknown UUIDs.
    # Returns response.Response({"vendors": [{"uuid": ..., "feature_ids": [...]}, ...]})

def lookup_features_by_subaccount_uuids(uuids):
    # Calls model, maps results back to input UUID order,
    # defaulting to feature_ids=[] for unknown UUIDs.
    # Returns response.Response({"subaccounts": [{"uuid": ..., "feature_ids": [...]}, ...]})
```

---

### Ticket 01c — ows-account: Handler + Schema Layer — PIP Endpoints

**Service**: ows-account | **Blocked by**: Ticket 01b (PP-1427) | **Blocks**: Ticket 07 (PP-1434)

#### Summary

Add request validation schemas and route handlers for the two new PIP endpoints. Both endpoints
follow the existing `NO access rule checks` PIP pattern used by other `/lookup/` routes.

#### Acceptance Criteria

- `POST /lookup/vendors/features/uuids/` exists and is reachable.
- `POST /lookup/subaccount/features/uuids/` exists and is reachable.
- Both endpoints return results as an ordered list matching the input UUID order (dataloader pattern).
- Unknown UUIDs are included in the response list with `feature_ids: []`.
- Request body is validated via marshmallow schemas before reaching the handler.
- Unit tests cover both endpoints (valid request, missing/invalid UUIDs).
- Integration tests in `tests/integration/api/test_vendor_lookups.py` cover both endpoints.

#### Implementation Details

**`account/validation/schemas/lookup.py`** — request schemas

```python
class LookupVendorFeaturesByUuids(Schema):
    uuids = fields.List(fields.UUID(required=True))

class LookupSubaccountFeaturesByUuids(Schema):
    uuids = fields.List(fields.UUID(required=True))
```

**`account/handlers/lookups.py`** — route handlers

```python
@app.route('/lookup/vendors/features/uuids/', methods=['POST'])
@validate_request_data(LookupVendorFeaturesByUuids())
def lookup_vendors_features_by_uuids(deserialize_schema):
    """NOTE: No access rule checks — PIP endpoint for Permission Platform."""
    uuids = deserialize_schema['uuids']
    return flaskify(feature.lookup_features_by_vendor_uuids(uuids))

@app.route('/lookup/subaccount/features/uuids/', methods=['POST'])
@validate_request_data(LookupSubaccountFeaturesByUuids())
def lookup_subaccount_features_by_uuids(deserialize_schema):
    """NOTE: No access rule checks — PIP endpoint for Permission Platform."""
    uuids = deserialize_schema['uuids']
    return flaskify(feature.lookup_features_by_subaccount_uuids(uuids))
```

#### Response Shapes

**Vendor endpoint** (`POST /lookup/vendors/features/uuids/`):
```json
{
  "vendors": [
    {"uuid": "uuid1", "feature_ids": [39, 43]},
    {"uuid": "uuid2", "feature_ids": []}
  ]
}
```

**Subaccount endpoint** (`POST /lookup/subaccount/features/uuids/`):
```json
{
  "subaccounts": [
    {"uuid": "subaccount_uuid1", "feature_ids": [39, 43]},
    {"uuid": "subaccount_uuid2", "feature_ids": []}
  ]
}
```

#### Integration Tests

**File**: `tests/integration/api/test_vendor_lookups.py`

These endpoints have no access rule checks (PIP pattern) so no bearer token is required — follow the same unauthenticated pattern used by the existing vendor/subaccount lookup tests in that file.

**Pre-requisite — QA test data**: Identify (or create) the following in the QA database before writing the tests:
- A vendor UUID whose parent vendor has at least one restricted feature (expected: non-empty `feature_ids` list in response).
  - A vendor UUID whose parent vendor has no restricted features (expected: `[]` in response).
  - A subaccount UUID whose parent vendor has at least one restricted feature (expected: non-empty list).
  - A subaccount UUID whose parent vendor has no restricted features (expected: `[]`).

Hardcode these UUIDs and their expected feature ID lists as constants at the top of the test module (matching the style of `vendor_id: 7123` / `uuid: '573d0372-...'` in the existing tests).

**Test cases**:

```python
def test_lookup_vendor_features_by_uuid_with_restricted_features():
    """POST /lookup/vendors/features/uuids/ returns enabled feature IDs for a vendor."""
    response = requests.post(
        url=f'{config.QA_BASE_URL}/lookup/vendors/features/uuids/',
        json={'uuids': [VENDOR_UUID_WITH_RESTRICTED_FEATURES]},
    )
    assert response.status_code == 200
    vendors = response.json()['vendors']
    assert vendors[0]['uuid'] == VENDOR_UUID_WITH_RESTRICTED_FEATURES
    assert vendors[0]['feature_ids'] == EXPECTED_FEATURE_IDS

def test_lookup_vendor_features_by_uuid_with_no_restricted_features():
    """POST /lookup/vendors/features/uuids/ returns [] for a vendor with no restrictions."""
    response = requests.post(
        url=f'{config.QA_BASE_URL}/lookup/vendors/features/uuids/',
        json={'uuids': [VENDOR_UUID_NO_RESTRICTIONS]},
    )
    assert response.status_code == 200
    vendors = response.json()['vendors']
    assert vendors[0]['uuid'] == VENDOR_UUID_NO_RESTRICTIONS
    assert vendors[0]['feature_ids'] == []

def test_lookup_vendor_features_by_uuid_unknown_uuid_returns_empty_feature_ids():
    """POST /lookup/vendors/features/uuids/ returns feature_ids=[] for an unknown UUID."""
    unknown_uuid = 'ffffffff-ffff-ffff-ffff-ffffffffffff'
    response = requests.post(
        url=f'{config.QA_BASE_URL}/lookup/vendors/features/uuids/',
        json={'uuids': [unknown_uuid]},
    )
    assert response.status_code == 200
    vendors = response.json()['vendors']
    assert vendors[0]['uuid'] == unknown_uuid
    assert vendors[0]['feature_ids'] == []

def test_lookup_subaccount_features_by_uuid_with_restricted_features():
    """POST /lookup/subaccount/features/uuids/ returns enabled feature IDs."""
    response = requests.post(
        url=f'{config.QA_BASE_URL}/lookup/subaccount/features/uuids/',
        json={'uuids': [SUBACCOUNT_UUID_WITH_RESTRICTED_FEATURES]},
    )
    assert response.status_code == 200
    subaccounts = response.json()['subaccounts']
    assert subaccounts[0]['uuid'] == SUBACCOUNT_UUID_WITH_RESTRICTED_FEATURES
    assert subaccounts[0]['feature_ids'] == EXPECTED_SUBACCOUNT_FEATURE_IDS

def test_lookup_subaccount_features_by_uuid_with_no_restricted_features():
    """POST /lookup/subaccount/features/uuids/ returns [] when parent vendor has no restrictions."""
    response = requests.post(
        url=f'{config.QA_BASE_URL}/lookup/subaccount/features/uuids/',
        json={'uuids': [SUBACCOUNT_UUID_NO_RESTRICTIONS]},
    )
    assert response.status_code == 200
    subaccounts = response.json()['subaccounts']
    assert subaccounts[0]['uuid'] == SUBACCOUNT_UUID_NO_RESTRICTIONS
    assert subaccounts[0]['feature_ids'] == []

def test_lookup_vendor_features_by_uuid_invalid_uuid_returns_400():
    """POST /lookup/vendors/features/uuids/ returns 400 for a malformed UUID."""
    response = requests.post(
        url=f'{config.QA_BASE_URL}/lookup/vendors/features/uuids/',
        json={'uuids': ['not-a-uuid']},
    )
    assert response.status_code == 400
```

---

### Ticket 02 — ows-pdp: CerbosPolicyParser + PolicyMetadataDatabase + Cache Key Constant

**Service**: ows-pdp | **Blocked by**: — | **Blocks**: Ticket 03 (PP-1430), Ticket 05 (PP-1432), Ticket 06 (PP-1433)

#### Summary

Create `pdp/connectors/cerbos_policy_parser.py` containing:

- `ResourcePolicyMetadata` — a Pydantic model capturing what a resource policy requires
  - `PolicyMetadataDatabase` — an in-memory dict-backed store with JSON serialization support
  - `CerbosPolicyParser` — walks the `cerbos/policies/` directory and builds the database

Also add the `CACHE_ENTRY_CERBOS_POLICY_METADATA` constant to `pdp/constants/constants.py`.

This module is the foundation for the entire feature-controls hydration flow. It is invoked by the
CI seed command (Ticket 03) and deserialized at Fargate startup (Ticket 05).

#### Acceptance Criteria

- `ResourcePolicyMetadata` has `resource_type: str` and `requires_account_feature_controls: bool`.
  - `PolicyMetadataDatabase.requires_account_feature_controls(resource_type)` returns `True` only for resource types whose policy YAML has `account_feature_controls` in `variables.import`.
  - `PolicyMetadataDatabase.to_json()` and `PolicyMetadataDatabase.from_json(raw)` round-trip correctly (serialize → deserialize → same data).
  - `CerbosPolicyParser.build_database()` recursively scans `*.yaml` / `*.yml`, skips non-resource policy files (no `resourcePolicy` key), and logs a warning (does not raise) for unparseable files.
  - Unit tests cover: happy path parse, missing `resourcePolicy` key skipped, bad YAML logged, `to_json`/`from_json` round-trip.
  - `CACHE_ENTRY_CERBOS_POLICY_METADATA = "cerbos_policy_metadata"` exists in constants and is added as a member of `CacheEntryType`.

#### Implementation Details

**New file: `pdp/connectors/cerbos_policy_parser.py`**

```python
import json
import logging
import yaml
from pathlib import Path
from typing import Dict, Optional
from pydantic import BaseModel

logger = logging.getLogger(__name__)


class ResourcePolicyMetadata(BaseModel):
    resource_type: str
    requires_account_feature_controls: bool


class PolicyMetadataDatabase:
    def __init__(self):
        self._policies: Dict[str, ResourcePolicyMetadata] = {}

    def add_policy(self, metadata: ResourcePolicyMetadata) -> None:
        self._policies[metadata.resource_type] = metadata

    def get_policy(self, resource_type: str) -> Optional[ResourcePolicyMetadata]:
        return self._policies.get(resource_type)

    def requires_account_feature_controls(self, resource_type: str) -> bool:
        policy = self.get_policy(resource_type)
        return policy.requires_account_feature_controls if policy else False

    def to_json(self) -> str:
        data = {rt: meta.model_dump() for rt, meta in self._policies.items()}
        return json.dumps(data)

    @classmethod
    def from_json(cls, raw: str) -> "PolicyMetadataDatabase":
        db = cls()
        for rt, meta_dict in json.loads(raw).items():
            db.add_policy(ResourcePolicyMetadata(**meta_dict))
        return db


class CerbosPolicyParser:
    def __init__(self, policies_dir: str = "cerbos/policies"):
        self.policies_dir = Path(policies_dir)

    def parse_resource_policy(self, policy_file: Path) -> Optional[ResourcePolicyMetadata]:
        try:
            with open(policy_file, "r") as f:
                policy_data = yaml.safe_load(f)
            if "resourcePolicy" not in policy_data:
                return None
            resource_policy = policy_data["resourcePolicy"]
            resource_type = resource_policy.get("resource")
            if not resource_type:
                logger.warning(f"No resource type found in {policy_file}")
                return None
            variables_import = resource_policy.get("variables", {}).get("import", [])
            return ResourcePolicyMetadata(
                resource_type=resource_type,
                requires_account_feature_controls="account_feature_controls" in variables_import,
            )
        except Exception as e:
            logger.error(f"Error parsing policy file {policy_file}: {e}")
            return None

    def build_database(self) -> PolicyMetadataDatabase:
        database = PolicyMetadataDatabase()
        policy_files = (
            list(self.policies_dir.rglob("*.yaml"))
            + list(self.policies_dir.rglob("*.yml"))
        )
        logger.info(f"Found {len(policy_files)} policy files to parse")
        for policy_file in policy_files:
            metadata = self.parse_resource_policy(policy_file)
            if metadata:
                database.add_policy(metadata)
        logger.info(
            f"Built policy database with {len(database._policies)} resource types"
        )
        return database
```

**Modified file: `pdp/constants/constants.py`**

```python
CACHE_ENTRY_CERBOS_POLICY_METADATA = "cerbos_policy_metadata"

class CacheEntryType(str, Enum):
    ...
    CACHE_ENTRY_CERBOS_POLICY_METADATA = CACHE_ENTRY_CERBOS_POLICY_METADATA
```

#### Notes

- `pyyaml` must be present in ows-pdp's dependencies. Verify before merging; add if missing.
  - `workstation_roles` detection is intentionally excluded from `ResourcePolicyMetadata` (deferred to PP-1411). Only `requires_account_feature_controls` is implemented here.
  - **Architecture**: The policy directory is crawled **once** in CI (Jenkinsfile), not at each Fargate task startup. The resulting `PolicyMetadataDatabase` is serialized to JSON and written to Redis. At startup, each Fargate task reads the entry from Redis, deserializes it, and injects it via the datasources/dependency-injection pattern. This mirrors the existing `bludgeon_cached_cerbos_decisions` CI pattern.

---

### Ticket 03 — ows-pdp: CLI Seed Command + Makefile Target

**Service**: ows-pdp | **Blocked by**: Ticket 02 (PP-1429) | **Blocks**: Ticket 04 (PP-1431), Ticket 05 (PP-1432)

#### Summary

Add a `seed_policy_metadata_cache` CLI command (invoked by Jenkins, not at Fargate task startup)
that crawls `cerbos/policies/`, builds a `PolicyMetadataDatabase`, serializes it to JSON, and
writes it to Redis under the `cerbos_policy_metadata` key.

Also add the corresponding `make seed_policy_metadata_cache` Makefile target.

This mirrors the existing `bludgeon_cached_cerbos_decisions` CI pattern.

#### Acceptance Criteria

- `./pdpcli cerbos seed_policy_metadata_cache` runs successfully and writes the JSON database to Redis under the `cerbos_policy_metadata` key.
  - `--policies-dir` option defaults to `"cerbos/policies"` and can be overridden.
  - `make seed_policy_metadata_cache` invokes the CLI command via the Docker wrapper.
  - The command prints the number of resource types written and the Redis key on success.
  - The new `cerbos` CLI group is registered in the main CLI app.
  - Manual smoke test: run the command locally against a local Redis instance and verify the key exists with correct content.

#### Implementation Details

**New file: `pdp/cli/commands/cerbos.py`**

```python
"""Commands to manage Cerbos policy metadata."""

import asyncio
import typer

from pdp import config
from pdp.connectors.cerbos_policy_parser import CerbosPolicyParser
from pdp.connectors.redis_client import RedisConnector
from pdp.constants.constants import CACHE_ENTRY_CERBOS_POLICY_METADATA

cli: typer.Typer = typer.Typer(
    short_help="Commands to manage Cerbos policy metadata", no_args_is_help=True
)


async def _seed_policy_metadata_cache(policies_dir: str) -> None:
    redis_connector = RedisConnector(
        config.REDIS_URL,
        use_redis_cache=config.CACHE_USE_REDIS,
    )
    parser = CerbosPolicyParser(policies_dir=policies_dir)
    db = parser.build_database()
    serialized = db.to_json()
    await redis_connector.set(key=CACHE_ENTRY_CERBOS_POLICY_METADATA, value=serialized)
    typer.secho(
        f"Wrote policy metadata database ({len(db._policies)} resource types) to "
        f"cache key '{CACHE_ENTRY_CERBOS_POLICY_METADATA}'.",
        fg="green",
    )


@cli.command("seed_policy_metadata_cache", short_help="Build and cache the policy metadata DB.")
def seed_policy_metadata_cache(
    policies_dir: str = typer.Option(
        "cerbos/policies",
        help="Path to the Cerbos policies directory.",
    ),
) -> None:
    """Crawl cerbos/policies, build PolicyMetadataDatabase, write to Redis."""
    asyncio.run(_seed_policy_metadata_cache(policies_dir))
```

**Modified file: `pdp/cli/main.py`** — register the new CLI group alongside existing groups:

```python
from pdp.cli.commands import cerbos as cerbos_commands
app.add_typer(cerbos_commands.cli, name="cerbos")
```

**Modified file: `Makefile`**

```makefile
PDPCLI_SEED_POLICY_METADATA_DOCKER_CMD = ./pdpcli cerbos seed_policy_metadata_cache

seed_policy_metadata_cache:
	${PDPCLI_SEED_POLICY_METADATA_DOCKER_CMD}
```

---

### Ticket 04 — ows-pdp: Jenkinsfile Seed Stages (QA + Prod)

**Service**: ows-pdp | **Blocked by**: Ticket 03 (PP-1430) | **Blocks**: —

#### Summary

Add two new Jenkinsfile stages — `Seed QA Policy Metadata Cache` and `Seed Prod Policy Metadata
Cache` — that invoke `make seed_policy_metadata_cache` after each environment's deploy stage.

Both stages are guarded by `isCerbosPolicyUpdated()` so they only run when Cerbos policy files
have changed. This mirrors the existing `Bludgeon *Cache` CI pattern.

#### Acceptance Criteria

- `Seed QA Policy Metadata Cache` stage runs on `master` branch when `isCerbosPolicyUpdated()` is `true`, using `QA_REDIS_URL`.
  - `Seed Prod Policy Metadata Cache` stage additionally requires `params.DEPLOY_TO_PROD == 'Yes'`, using `PROD_REDIS_URL`.
  - Both stages are placed immediately after (or alongside) the corresponding `Bludgeon *Cache` stages.
  - When a policy file changes and the pipeline runs successfully, a manual check confirms the Redis key `cerbos_policy_metadata` is updated in both QA and Prod Redis.

#### Implementation Details

**Modified file: `Jenkinsfile`** — add after the QA `Bludgeon Cache` stage:

```groovy
stage('Seed QA Policy Metadata Cache') {
    when {
        allOf {
            branch 'master'
            expression { isCerbosPolicyUpdated() == true }
        }
    }
    steps {
        withEcr {
            withEnv([
                "REDIS_URL=${QA_REDIS_URL}",
                "CACHE_USE_REDIS=${CACHE_USE_REDIS}"
            ]) {
                withAWS(role: QA_DEPLOYMENT_ROLE, roleAccount: QA_ACCOUNT_ID, roleSessionName: SESSION_NAME, useNode: true) {
                    sh 'make seed_policy_metadata_cache'
                }
            }
        }
    }
}
```

Add after the Prod `Bludgeon Cache` stage:

```groovy
stage('Seed Prod Policy Metadata Cache') {
    when {
        allOf {
            branch 'master'
            expression { isCerbosPolicyUpdated() == true }
            expression { params.DEPLOY_TO_PROD == 'Yes' }
        }
    }
    steps {
        withEcr {
            withEnv([
                "REDIS_URL=${PROD_REDIS_URL}",
                "CACHE_USE_REDIS=${CACHE_USE_REDIS}"
            ]) {
                withAWS(role: PROD_DEPLOYMENT_ROLE, roleAccount: PROD_ACCOUNT_ID, roleSessionName: SESSION_NAME, useNode: true) {
                    sh 'make seed_policy_metadata_cache'
                }
            }
        }
    }
}
```

#### Notes

- The seed stages run **after** the corresponding deploy stage so the newly deployed image (with updated policies already baked in) is what Jenkins reads from the filesystem during the seed.
  - If the seed stage is skipped (e.g., no policy changes), the existing Redis entry remains valid from the previous run.

---

### Ticket 05 — ows-pdp: Read PolicyMetadataDatabase from Redis at Startup

**Service**: ows-pdp | **Blocked by**: Ticket 02 (PP-1429), Ticket 03 (PP-1430) | **Blocks**: Ticket 09a (PP-1436)

#### Summary

Update `pdp/fastapi/datasources.py` so that each Fargate task reads the pre-seeded
`cerbos_policy_metadata` Redis entry at startup, deserializes it into a `PolicyMetadataDatabase`
instance, and makes it available via the standard dependency-injection pattern.

Fail loudly (raise `RuntimeError`) if the key is missing — a missing key indicates a misconfigured
deployment and should be caught immediately rather than silently degrading.

#### Acceptance Criteria

- On startup, the task reads `cerbos_policy_metadata` from Redis and deserializes it.
  - If the Redis key is missing, startup raises `RuntimeError` with a helpful message.
  - `get_policy_metadata_database()` is available as a FastAPI dependency injector.
  - `DATA_SOURCES[POLICY_METADATA_DATABASE_KEY]` holds the `PolicyMetadataDatabase` instance.
  - Log line at startup reports how many resource types were loaded.

#### Implementation Details

**Modified file: `pdp/fastapi/datasources.py`**

```python
# Add to imports
from pdp.connectors.cerbos_policy_parser import PolicyMetadataDatabase
from pdp.constants.constants import CACHE_ENTRY_CERBOS_POLICY_METADATA

# New constant
POLICY_METADATA_DATABASE_KEY = "POLICY_METADATA_DATABASE"


# Inside datasources_lifespan, after redis_connector is available:
raw = await redis_connector.get(key=CACHE_ENTRY_CERBOS_POLICY_METADATA)
if raw is None:
    raise RuntimeError(
        f"Cache entry '{CACHE_ENTRY_CERBOS_POLICY_METADATA}' not found. "
        "Run 'make seed_policy_metadata_cache' (or the Jenkins stage) to populate it."
    )
policy_metadata_db = PolicyMetadataDatabase.from_json(raw)
logger.info(
    "[lifespan] Loaded policy metadata database from cache "
    f"({len(policy_metadata_db._policies)} resource types)"
)
DATA_SOURCES[POLICY_METADATA_DATABASE_KEY] = policy_metadata_db


# New dependency injector
def get_policy_metadata_database() -> PolicyMetadataDatabase:
    """Dependency injector for the shared PolicyMetadataDatabase."""
    database = DATA_SOURCES[POLICY_METADATA_DATABASE_KEY]
    assert isinstance(database, PolicyMetadataDatabase)
    return database
```

#### Notes

- The `redis_connector` must be initialized before this block runs — place the read after the existing Redis setup in `datasources_lifespan`.
  - No filesystem access at startup: the Fargate task never reads `cerbos/policies/` directly. That happens only in CI via the seed command.

---

### Ticket 06 — ows-pdp: Add needs_account_feature_controls_lookup to Resource

**Service**: ows-pdp | **Blocked by**: Ticket 02 (PP-1429) | **Blocks**: Ticket 09a (PP-1436)

#### Summary

Implement the `needs_account_feature_controls_lookup` stub method on the `Resource` class in
`pdp/fastapi/schemas/identity.py`. The method delegates to `PolicyMetadataDatabase` to determine
whether the resource's policy imports `account_feature_controls`.

#### Acceptance Criteria

- `resource.needs_account_feature_controls_lookup(policy_db)` returns `True` when `policy_db.requires_account_feature_controls(resource.resource_type)` is `True`.
  - Returns `False` for resource types not present in the database.
  - Unit tests cover both branches.

#### Implementation Details

**Modified file: `pdp/fastapi/schemas/identity.py`**

```python
from pdp.connectors.cerbos_policy_parser import PolicyMetadataDatabase  # TYPE_CHECKING import

class Resource(BaseModel):
    resource_id: Union[Annotated[str, StringConstraints(min_length=1)], int]
    resource_type: Annotated[str, StringConstraints(min_length=1)]
    attributes: Dict[str, Any] = {}

    model_config = ConfigDict(str_strip_whitespace=True)

    def needs_account_feature_controls_lookup(
        self,
        policy_db: "PolicyMetadataDatabase",
    ) -> bool:
        """Return True if this resource type's policy imports account_feature_controls."""
        return policy_db.requires_account_feature_controls(self.resource_type)
```

#### Notes

- Use `TYPE_CHECKING` guard on the import to avoid circular imports if necessary.
  - The existing stub(s) on `Resource` for this purpose should be replaced, not duplicated.

---

### Ticket 07 — ows-pdp: OwsAccountClient Response Models + Client Methods

**Service**: ows-pdp | **Blocked by**: Ticket 01c (PP-1428) | **Blocks**: Ticket 08 (PP-1435)

#### Summary

Extend `pdp/connectors/ows_account.py` with the Pydantic response models and async client methods
needed to call the two new ows-account feature-controls endpoints added in Ticket 01.

#### Acceptance Criteria

- `LookupVendorFeaturesResponse` and `LookupSubaccountFeaturesResponse` Pydantic models exist and parse the expected response shapes.
  - `OwsAccountClient.get_features_for_vendor_uuids(uuids)` calls `POST /lookup/vendors/features/uuids/` and returns a `LookupVendorFeaturesResponse`.
  - `OwsAccountClient.get_features_for_subaccount_uuids(uuids)` calls `POST /lookup/subaccount/features/uuids/` and returns a `LookupSubaccountFeaturesResponse`.
  - Both methods call `raise_for_status()` on the HTTP response before parsing.
  - Unit tests cover model parsing (valid response, missing UUIDs, empty feature list).

#### Implementation Details

**Modified file: `pdp/connectors/ows_account.py`**

```python
class LookupVendorFeatures(BaseModel):
    uuid: UUID
    feature_ids: List[int]

class LookupSubaccountFeatures(BaseModel):
    uuid: UUID
    feature_ids: List[int]

class LookupVendorFeaturesResponse(BaseModel):
    vendors: List[LookupVendorFeatures]

class LookupSubaccountFeaturesResponse(BaseModel):
    subaccounts: List[LookupSubaccountFeatures]


# On OwsAccountClient:

async def get_features_for_vendor_uuids(
    self, uuids: List[UUID]
) -> LookupVendorFeaturesResponse:
    response = await self.async_ows_client.post(
        self.service_name,
        path="/lookup/vendors/features/uuids/",
        json={"uuids": [str(u) for u in uuids]},
    )
    response.raise_for_status()
    return LookupVendorFeaturesResponse.model_validate_json(response.content)

async def get_features_for_subaccount_uuids(
    self, uuids: List[UUID]
) -> LookupSubaccountFeaturesResponse:
    response = await self.async_ows_client.post(
        self.service_name,
        path="/lookup/subaccount/features/uuids/",
        json={"uuids": [str(u) for u in uuids]},
    )
    response.raise_for_status()
    return LookupSubaccountFeaturesResponse.model_validate_json(response.content)
```

---

### Ticket 08 — ows-pdp: VendorFeatureControlsProxy

**Service**: ows-pdp | **Blocked by**: Ticket 07 (PP-1434) | **Blocks**: Ticket 09a (PP-1436)

#### Summary

Create `pdp/proxies/vendor_feature_controls_proxy.py` containing `VendorFeatureControlsProxy`.

The proxy accepts lists of account and subaccount UUIDs, calls the two ows-account feature
endpoints **concurrently** via `asyncio.gather`, and exposes a `get_feature_controls(uuid)`
accessor. Keeping the fetch logic here isolates it from the hydration function and makes it
independently testable.

This is modelled on `UuidToIdExchangeTenantProxy` in the existing proxies directory.

#### Acceptance Criteria

- `gather_feature_controls()` calls both ows-account endpoints concurrently and merges results.
  - `get_feature_controls(uuid)` returns the feature ID list for a given tenant UUID, or `[]` if not found.
  - Accessing `gathered_feature_controls` before calling `gather_feature_controls()` raises `FeatureControlsLookupError`.
  - If `account_uuids` is empty, the vendor endpoint is not called (and vice versa for `subaccount_uuids`).
  - Unit tests cover: concurrent fetch, empty UUID lists, UUID not in response returns `[]`.

#### Implementation Details

**New file: `pdp/proxies/vendor_feature_controls_proxy.py`**

```python
"""VendorFeatureControlsProxy."""

import asyncio
import logging
from typing import Dict, List, Optional
from uuid import UUID

from ddtrace import tracer

from pdp.connectors.ows_account import (
    LookupSubaccountFeaturesResponse,
    LookupVendorFeaturesResponse,
    OwsAccountClient,
)
from pdp.proxies.helpers import lookup_with_error_handling

logger = logging.getLogger(__name__)


class FeatureControlsLookupError(Exception):
    """Raised when a feature controls lookup fails."""
    pass


class VendorFeatureControlsProxy:
    """Fetches vendor feature controls for account and subaccount tenants.

    Calls two dedicated ows-account endpoints concurrently:
      - POST /lookup/vendors/features/uuids/     (account tenants)
      - POST /lookup/subaccount/features/uuids/  (subaccount tenants)

    ows-account handles UUID→vendor_id resolution internally for both endpoints.
    """

    def __init__(
        self,
        account_uuids: List[UUID],
        subaccount_uuids: List[UUID],
        ows_account_client: OwsAccountClient,
    ):
        self._account_uuids = list(set(account_uuids))
        self._subaccount_uuids = list(set(subaccount_uuids))
        self._ows_account_client = ows_account_client
        self._gathered_feature_controls: Optional[Dict[UUID, List[int]]] = None

    @property
    def gathered_feature_controls(self) -> Dict[UUID, List[int]]:
        if self._gathered_feature_controls is None:
            raise FeatureControlsLookupError(
                "Attempted to access feature controls before ows-account lookups."
            )
        return self._gathered_feature_controls

    @tracer.wrap()
    def get_feature_controls(self, uuid: UUID) -> List[int]:
        """Return feature control IDs for a tenant UUID. Returns [] if not found."""
        return self.gathered_feature_controls.get(uuid, [])

    @tracer.wrap()
    async def _fetch_vendor_feature_controls(self) -> Dict[UUID, List[int]]:
        if not self._account_uuids:
            return {}
        response: LookupVendorFeaturesResponse = await lookup_with_error_handling(
            lookup_cb=self._ows_account_client.get_features_for_vendor_uuids(
                uuids=self._account_uuids,
            ),
            http_error_message_400_string="Bad request to ows-account for vendor feature controls lookup.",
            http_error_message_500_string="Vendor feature controls lookup by vendor_uuids failed.",
            unhandled_exception_string="Server error: vendor feature controls lookup by vendor_uuids failed.",
            lookup_error_cls=FeatureControlsLookupError,
        )
        return {item.uuid: item.feature_ids for item in response.vendors}

    @tracer.wrap()
    async def _fetch_subaccount_feature_controls(self) -> Dict[UUID, List[int]]:
        if not self._subaccount_uuids:
            return {}
        response: LookupSubaccountFeaturesResponse = await lookup_with_error_handling(
            lookup_cb=self._ows_account_client.get_features_for_subaccount_uuids(
                uuids=self._subaccount_uuids,
            ),
            http_error_message_400_string="Bad request to ows-account for subaccount feature controls lookup.",
            http_error_message_500_string="Subaccount feature controls lookup by subaccount_uuids failed.",
            unhandled_exception_string="Server error: subaccount feature controls lookup by subaccount_uuids failed.",
            lookup_error_cls=FeatureControlsLookupError,
        )
        return {item.uuid: item.feature_ids for item in response.subaccounts}

    @tracer.wrap()
    async def gather_feature_controls(self) -> Dict[UUID, List[int]]:
        """Fetch feature controls for all account and subaccount tenants concurrently."""
        self._gathered_feature_controls = {}
        vendor_controls, subaccount_controls = await asyncio.gather(
            self._fetch_vendor_feature_controls(),
            self._fetch_subaccount_feature_controls(),
        )
        self._gathered_feature_controls.update(vendor_controls)
        self._gathered_feature_controls.update(subaccount_controls)
        return self._gathered_feature_controls
```

---

### Ticket 09a — ows-pdp: Logic Layer — Hydration Function + check_resources Integration

**Service**: ows-pdp | **Blocked by**: Ticket 05 (PP-1432), Ticket 06 (PP-1433), Ticket 08 (PP-1435) | **Blocks**: Ticket 09b (PP-1437)

#### Summary

Add `_hydrate_resources_with_feature_controls_as_needed` to `pdp/logic/cerbos.py` and call it
inside `check_resources` before the Cerbos call. After this ticket the hydration logic is complete
and tested, but not yet wired to any endpoint (that is Ticket 09b).

#### Acceptance Criteria

- `_hydrate_resources_with_feature_controls_as_needed` exists and hydrates only resources whose policy imports `account_feature_controls`.
  - Resources that don't need feature controls are untouched (no extra HTTP calls made).
  - `check_resources` accepts a `policy_metadata_db` parameter and calls the hydration function before the Cerbos call.
  - Unit/integration tests cover: resource needing feature controls gets them injected; resource not needing feature controls is unaffected; tenant not found in ows-account response results in `[]`.

#### Implementation Details

**Modified file: `pdp/logic/cerbos.py`** — new hydration function

```python
@tracer.wrap()
async def _hydrate_resources_with_feature_controls_as_needed(
    check_resources_request: CheckResourcesRequest,
    policy_metadata_db: PolicyMetadataDatabase,
    ows_account_client: OwsAccountClient,
) -> CheckResourcesRequest:
    """Hydrate resources with account feature controls where the policy requires it.

    Pass 1: collect tenant UUIDs by type for resources that need feature controls.
    Pass 2: fetch feature controls for all UUIDs concurrently via VendorFeatureControlsProxy.
    Pass 3: inject feature IDs into resource.attributes["tenant"]["account_feature_controls"].
    """
    # Pass 1
    account_uuids: set[UUID] = set()
    subaccount_uuids: set[UUID] = set()
    for cra in check_resources_request.resources:
        if cra.resource.needs_account_feature_controls_lookup(policy_metadata_db):
            tenant = cra.get_tenant()
            if tenant:
                if tenant.tenant_type == TenantType.ACCOUNT:
                    account_uuids.add(tenant.tenant_uuid)
                elif tenant.tenant_type == TenantType.SUBACCOUNT:
                    subaccount_uuids.add(tenant.tenant_uuid)

    if not account_uuids and not subaccount_uuids:
        return check_resources_request

    # Pass 2
    proxy = VendorFeatureControlsProxy(
        account_uuids=list(account_uuids),
        subaccount_uuids=list(subaccount_uuids),
        ows_account_client=ows_account_client,
    )
    await proxy.gather_feature_controls()

    # Pass 3
    for cra in check_resources_request.resources:
        resource = cra.resource
        if not resource.needs_account_feature_controls_lookup(policy_metadata_db):
            continue
        tenant = cra.get_tenant()
        if not tenant or tenant.tenant_type not in (TenantType.ACCOUNT, TenantType.SUBACCOUNT):
            continue
        feature_ids = proxy.get_feature_controls(tenant.tenant_uuid)
        if "tenant" not in resource.attributes:
            resource.attributes["tenant"] = {}
        resource.attributes["tenant"]["account_feature_controls"] = feature_ids

    return check_resources_request
```

**Modified file: `pdp/logic/cerbos.py`** — updated `check_resources` signature

```python
async def check_resources(
    identity_uuid: str,
    check_resources_request: CheckResourcesRequest,
    cerbos_client: AsyncCerbosClient,
    pdp_tenant_roles: Dict[uuid.UUID, TenantRoles],
    ows_account_client: OwsAccountClient,
    ows_participant_client: OwsParticipantClient,
    redis_connector: RedisConnector,
    policy_metadata_db: PolicyMetadataDatabase,          # NEW
    ows_permissions_tenant_roles: Optional[Dict[uuid.UUID, TenantRoles]] = None,
    include_resource_attributes: bool = False,
    authenticated_identity_uuid: Optional[UUID4] = None,
    principal: Optional[Principal] = None,
) -> CheckResourcesResponse:
    # ... existing principal build + hierarchy hydration ...

    # NEW: hydrate feature controls
    check_resources_request = await _hydrate_resources_with_feature_controls_as_needed(
        check_resources_request=check_resources_request,
        policy_metadata_db=policy_metadata_db,
        ows_account_client=ows_account_client,
    )

    # existing Cerbos call
    result = await _paginated_resource_check(...)
    return result
```

#### Notes

- **Hydration targets**: `account_feature_controls` affects the **Resource** (`resource.attributes.tenant.account_feature_controls`). `workstation_roles` (deferred to PP-1411) affects the **Principal** (derived roles via `importDerivedRoles`). These are separate hydration paths targeting different Cerbos inputs.
  - Feature controls are **not cached** per-request or across requests. This follows the same rationale as `get_principal_ows_permissions_from_scope` which also does not cache. See the spike doc for details.
  - If a tenant UUID is absent from the ows-account response, `proxy.get_feature_controls(uuid)` returns `[]`, which is a safe default (no features enabled means most restrictive policy applies).

---

### Ticket 09b — ows-pdp: Handler Layer — Endpoint Injection

**Service**: ows-pdp | **Blocked by**: Ticket 09a (PP-1436) | **Blocks**: —

#### Summary

Inject `policy_metadata_db` as a FastAPI dependency in all router endpoints that call
`check_resources`. This is a targeted wiring change in `pdp/fastapi/routers/identity.py` —
the logic itself is complete in Ticket 09a; this ticket simply plumbs the dependency through
to the call sites.

After this ticket the full feature-controls hydration flow is functional end-to-end.

#### Acceptance Criteria

- `check_my_resources`, `check_identity_resources`, and any other endpoints calling `check_resources` declare `policy_metadata_db: PolicyMetadataDatabase = Depends(get_policy_metadata_database)`.
  - Each of those endpoints passes `policy_metadata_db=policy_metadata_db` to `cerbos.check_resources(...)`.
  - End-to-end smoke test: a request for a resource type whose policy imports `account_feature_controls` results in the correct feature IDs in `resource.attributes["tenant"]["account_feature_controls"]` before the Cerbos call.
  - Integration tests in `tests/integration/api/identity/test_check_my_resources.py` cover the feature-controls hydration path end-to-end.

#### Implementation Details

**Modified file: `pdp/fastapi/routers/identity.py`**

```python
from pdp.fastapi.datasources import (
    ...,
    get_policy_metadata_database,  # NEW
)

# In check_my_resources, check_identity_resources, and any other endpoint calling check_resources:
async def check_my_resources(
    ...,
    policy_metadata_db: PolicyMetadataDatabase = Depends(get_policy_metadata_database),  # NEW
) -> ...:
    ...
    check_response = await cerbos.check_resources(
        ...,
        policy_metadata_db=policy_metadata_db,  # NEW
    )
```

#### Integration Tests

**File**: `tests/integration/api/identity/test_check_my_resources.py`

**Pre-requisites — QA test data and credentials**:

- Identify (or create) a vendor UUID in QA that has at least one restricted feature. This vendor UUID is used as the `tenant_uuid` in test requests.
  - Identify (or create) a subaccount UUID in QA whose parent vendor also has restricted features.
  - Create a new integration test user in Auth0 (QA) that has a role on the above vendor and subaccount. Store credentials in AWS Secrets Manager following the existing `SecretLookupInfo` pattern. Add a new fixture to `tests/integration/conftest.py`:

```python
@pytest.fixture(scope="session")
def bearer_token_vendor_features_test_user(
    generate_bearer_token: Callable[..., str],
    jwtauth_secrets_manager: JwtAuthSecretsManager,
) -> str:
    """Bearer token for integration test user scoped to a vendor with feature controls."""
    return generate_bearer_token(
        get_user_creds_args=SecretLookupInfo(
            environment="qa",
            service_name=utils.APPLICATION,
            secret_name=utils.PP_VENDOR_FEATURES_TEST_USER_CREDENTIALS,
        ),
        get_auth0_creds_args=SecretLookupInfo(
            environment="qa",
            service_name=utils.APPLICATION,
            secret_name=utils.PDP_TEST_APP_AUTH0_CREDENTIALS,
        ),
        secrets_manager=jwtauth_secrets_manager,
    )

@pytest.fixture(scope="session")
def bearer_token_vendor_features_test_user_identity_uuid(
    bearer_token_vendor_features_test_user: str,
) -> Optional[str]:
    """Extract identity_uuid from vendor features test user bearer token."""
    return utils.get_bearer_token_identity_uuid(bearer_token_vendor_features_test_user)
```

**Test cases** — use `include_resource_attributes_in_response: True` so the hydrated attributes are visible in the response:

```python
def test_check_my_resources_injects_account_feature_controls_for_vendor(
    default_boto_client: DynamoDBClient,
    bearer_token_vendor_features_test_user: str,
    bearer_token_vendor_features_test_user_identity_uuid: str,
) -> None:
    """Feature controls are injected into resource attributes for a resource type
    whose policy imports account_feature_controls, when the identity is an account tenant."""
    seed_test_pp_identity(
        default_boto_client,
        bearer_token_vendor_features_test_user_identity_uuid,
        VENDOR_UUID_WITH_RESTRICTED_FEATURES,
        "<role>",
        tenant_type="account",
    )
    body = {
        "resources": [
            {
                "resource": {
                    "resource_id": "1",
                    "resource_type": "<resource_type_that_imports_account_feature_controls>",
                    "attributes": {
                        "tenant": {
                            "tenant_type": "account",
                            "tenant_uuid": VENDOR_UUID_WITH_RESTRICTED_FEATURES,
                        }
                    },
                },
                "action": "<action>",
            }
        ],
        "include_resource_attributes_in_response": True,
    }
    response = requests.post(
        f"{config.QA_BASE_URL}/identity/self/check/resources/",
        json=body,
        headers={"Authorization": f"Bearer {bearer_token_vendor_features_test_user}"},
    )
    assert response.status_code == 200
    resource_attrs = response.json()["resources"][0]["resource"]["attributes"]
    assert resource_attrs["tenant"]["account_feature_controls"] == EXPECTED_FEATURE_IDS


def test_check_my_resources_injects_account_feature_controls_for_subaccount(
    default_boto_client: DynamoDBClient,
    bearer_token_vendor_features_test_user: str,
    bearer_token_vendor_features_test_user_identity_uuid: str,
) -> None:
    """Feature controls are injected for a subaccount tenant."""
    seed_test_pp_identity(
        default_boto_client,
        bearer_token_vendor_features_test_user_identity_uuid,
        SUBACCOUNT_UUID_WITH_RESTRICTED_FEATURES,
        "<role>",
        tenant_type="subaccount",
    )
    body = {
        "resources": [
            {
                "resource": {
                    "resource_id": "1",
                    "resource_type": "<resource_type_that_imports_account_feature_controls>",
                    "attributes": {
                        "tenant": {
                            "tenant_type": "subaccount",
                            "tenant_uuid": SUBACCOUNT_UUID_WITH_RESTRICTED_FEATURES,
                        }
                    },
                },
                "action": "<action>",
            }
        ],
        "include_resource_attributes_in_response": True,
    }
    response = requests.post(
        f"{config.QA_BASE_URL}/identity/self/check/resources/",
        json=body,
        headers={"Authorization": f"Bearer {bearer_token_vendor_features_test_user}"},
    )
    assert response.status_code == 200
    resource_attrs = response.json()["resources"][0]["resource"]["attributes"]
    assert resource_attrs["tenant"]["account_feature_controls"] == EXPECTED_SUBACCOUNT_FEATURE_IDS


def test_check_my_resources_does_not_inject_feature_controls_for_unrelated_resource_type(
    default_boto_client: DynamoDBClient,
    bearer_token_pdptest_user: str,
    bearer_token_pdptest_user_identity_uuid: str,
) -> None:
    """Feature controls are NOT injected for resource types that do not import account_feature_controls."""
    seed_test_pp_identity(
        default_boto_client,
        bearer_token_pdptest_user_identity_uuid,
        "42879e8c-9f47-4214-b611-1e6feb0be6af",
        "audience_development_analyst",
        tenant_type="account",
    )
    body = {
        "resources": [
            {
                "resource": {
                    "resource_id": "1",
                    "resource_type": "fan_data_list",  # does not import account_feature_controls
                    "attributes": {
                        "tenant": {
                            "tenant_type": "account",
                            "tenant_uuid": "42879e8c-9f47-4214-b611-1e6feb0be6af",
                        }
                    },
                },
                "action": "view",
            }
        ],
        "include_resource_attributes_in_response": True,
    }
    response = requests.post(
        f"{config.QA_BASE_URL}/identity/self/check/resources/",
        json=body,
        headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"},
    )
    assert response.status_code == 200
    resource_attrs = response.json()["resources"][0]["resource"]["attributes"]
    assert "account_feature_controls" not in resource_attrs.get("tenant", {})
```

> **Note**: The `<resource_type_that_imports_account_feature_controls>`, `<role>`, `<action>`, and UUID constants above are placeholders. Fill them in once the first Cerbos policy that uses `account_feature_controls` is written (likely as part of PP-1388 or the first consuming ticket). The test structure and assertion pattern will not change.

---

### Ticket 10 — ows-pdp: Feature Flag `pp_vendor_features_lookup`

**Service**: ows-pdp | **Blocked by**: — | **Blocks**: Ticket 05 (PP-1432), Ticket 09a (PP-1436)

#### Summary

Add the `FEATURE_FLAG_PP_VENDOR_FEATURES_LOOKUP` constant to `pdp/constants/features.py` and
create the corresponding `pp_vendor_features_lookup` flag in Split.io (defaulting to OFF).

This flag gates the vendor feature controls hydration in `check_resources` (Ticket 09a): if the
flag is OFF for an identity, `_hydrate_resources_with_feature_controls_as_needed` is skipped
entirely. This decouples the code deployment from the traffic rollout and allows Ticket 05's
startup Redis read to soft-fail gracefully (log warning instead of `RuntimeError`) when the flag
is still off, since no request will actually reach the hydration path.

Tickets 05 and 09a both reference this constant, so it must land first.

#### Acceptance Criteria

- `FEATURE_FLAG_PP_VENDOR_FEATURES_LOOKUP = "pp_vendor_features_lookup"` exists in `pdp/constants/features.py`.
  - Flag `pp_vendor_features_lookup` exists in Split.io and defaults to OFF.
  - No other code changes beyond the constant definition — the usages are implemented in Tickets 05 and 09a.

#### Implementation Details

**Modified file: `pdp/constants/features.py`**

```python
FEATURE_FLAG_PP_VENDOR_FEATURES_LOOKUP = "pp_vendor_features_lookup"
```

**Manual step**: Create flag `pp_vendor_features_lookup` in Split.io. Default treatment: OFF.

#### Notes

- The flag is identity-scoped (checked via `BooleanFeature.is_on_for_identity(str(identity_uuid))`), consistent with `FEATURE_FLAG_PP_SEND_IMPERSONATED_BY_IDENTITY_UUID`.
  - The flag check itself lives in the Cerbos logic layer (`pdp/logic/cerbos.py`), not in the router, so the router doesn't need to branch on it.

---

### Ticket 11 — ows-pdp: Infra Endpoint — Tenant Feature Controls Lookup

**Service**: ows-pdp | **Blocked by**: Ticket 08 (PP-1435) | **Blocks**: —

#### Summary

Add a new infra endpoint `POST /infra/tenant/feature-controls/` that accepts a list of `Tenant`
objects (each with `tenant_uuid` and `tenant_type`) and returns a dictionary mapping each UUID to
its enabled feature control IDs.

This follows the same infra pattern as `POST /infra/tenant/uuid-to-id-exchange/` — authenticated
via `check_authorization_infra`, logic in `pdp/logic/infra.py`, and proxy-delegated to
`VendorFeatureControlsProxy` from Ticket 08. It is intended as a debugging and verification
endpoint for the feature-controls hydration pipeline, not a hot-path production endpoint.

#### Acceptance Criteria

- `POST /infra/tenant/feature-controls/` accepts a JSON array of `Tenant` objects and returns
  `Dict[UUID, TenantFeatureControls]`.
- `TenantFeatureControls` schema has a single field: `account_feature_controls: List[int]`.
- Account and subaccount tenants are dispatched to the correct ows-account endpoint; tenants with
  other `tenant_type` values return `account_feature_controls: []` and are not sent upstream.
- A UUID present in the request but absent from the ows-account response returns
  `account_feature_controls: []` (safe default, consistent with `proxy.get_feature_controls`).
- Endpoint is protected by `check_authorization_infra` (same as existing infra endpoints).
- Unit tests cover: mixed account/subaccount input, unknown UUID returns `[]`, empty input returns
  `{}`, non-account/subaccount tenant type is excluded.

#### Implementation Details

**Modified file: `pdp/fastapi/schemas/tenant.py`** — new response model

```python
class TenantFeatureControls(BaseModel):
    """Feature controls for a single tenant, keyed by UUID in the endpoint response."""

    account_feature_controls: List[int]
```

**Modified file: `pdp/config.py`** — new URL constant

```python
TENANT_FEATURE_CONTROLS = "/infra/tenant/feature-controls/"
```

**Modified file: `pdp/logic/infra.py`** — new logic function

```python
from pdp.fastapi.schemas.tenant import Tenant, TenantFeatureControls
from pdp.proxies.vendor_feature_controls_proxy import VendorFeatureControlsProxy

async def gather_tenant_feature_controls(
    tenants: List[Tenant],
    ows_account_client: OwsAccountClient,
) -> Dict[uuid.UUID, TenantFeatureControls]:
    """Return feature controls for each tenant UUID via VendorFeatureControlsProxy."""
    account_uuids = [
        t.tenant_uuid for t in tenants if t.tenant_type == TenantType.ACCOUNT
    ]
    subaccount_uuids = [
        t.tenant_uuid for t in tenants if t.tenant_type == TenantType.SUBACCOUNT
    ]
    proxy = VendorFeatureControlsProxy(
        account_uuids=account_uuids,
        subaccount_uuids=subaccount_uuids,
        ows_account_client=ows_account_client,
    )
    await proxy.gather_feature_controls()
    return {
        t.tenant_uuid: TenantFeatureControls(
            account_feature_controls=proxy.get_feature_controls(t.tenant_uuid),
        )
        for t in tenants
    }
```

**Modified file: `pdp/fastapi/routers/infra.py`** — new endpoint

```python
@router.post(
    config.TENANT_FEATURE_CONTROLS,
    description="Return enabled feature control IDs for a list of tenant UUIDs",
    response_model=Dict[UUID, tenant_schema.TenantFeatureControls],
    dependencies=[Depends(check_authorization_infra)],
)
async def tenant_feature_controls(
    tenants: List[tenant_schema.Tenant],
    ows_account_client: OwsAccountClient = Depends(get_ows_account_client),
) -> Dict[UUID, tenant_schema.TenantFeatureControls]:
    """Look up vendor feature controls for each tenant in the request list."""
    return await infra.gather_tenant_feature_controls(
        tenants=tenants,
        ows_account_client=ows_account_client,
    )
```

#### Request / Response Shapes

**Request** (`POST /infra/tenant/feature-controls/`):
```json
[
    {"tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account"},
    {"tenant_uuid": "0c54f0f1-deda-428a-a00c-9317170db544", "tenant_type": "subaccount"}
]
```

**Response**:
```json
{
    "573d0372-7f2f-48a6-8deb-c9a6558f9549": {"account_feature_controls": [39, 43]},
    "0c54f0f1-deda-428a-a00c-9317170db544": {"account_feature_controls": []}
}
```

#### Notes

- The `redis_connector` dependency is **not** needed here — `VendorFeatureControlsProxy` calls
  ows-account directly and does not use the tenant-hierarchy cache.
- Tenants with `tenant_type` values other than `account` or `subaccount` (e.g., `company_brand`)
  are included in the response with `account_feature_controls: []` but are not forwarded to
  ows-account. This matches the proxy's scope, which only supports account and subaccount lookups.
- No feature flag guard on this endpoint — it is an infra utility that bypasses the
  `check_resources` hot path entirely.

---

### Ticket 12 — ows-account: Redis Cache + Invalidation for Vendor Features Lookup Endpoints

**Service**: ows-account | **Blocked by**: Ticket 01c (PP-1428) | **Blocks**: —

#### Summary

ows-account has no Redis infrastructure today. This ticket provisions ElastiCache (QA and Prod
Terraform), creates the Redis connector, and wires a per-UUID cache-aside layer into the two
feature-controls lookup handlers added in Ticket 01c. The existing restricted-features write
paths (`bulk_add` / `bulk_remove`) are updated to invalidate affected cache entries on success.

#### Acceptance Criteria

- `account/connectors/redis.py` exists with `get`, `set`, `delete`, and `delete_keys` functions; falls back to `fakeredis` when `REDIS_URL` is unset.
- `account/config.py` exposes `REDIS_URL` (default `None`) and `REDIS_CACHE_TTL` (default `3600`).
- `POST /lookup/vendors/features/uuids/` reads per-UUID entries from cache before querying the DB; writes fetched results back to cache; assembles the response in input UUID order.
- `POST /lookup/subaccount/features/uuids/` follows the same per-UUID cache-aside pattern.
- `bulk_add_restricted_features_for_vendor` deletes cache entries for the affected vendor UUID and all subaccount UUIDs belonging to that vendor after a successful write. Invalidation failure logs a warning and does not surface to the caller.
- `bulk_remove_restricted_features_for_vendor` applies the same invalidation.
- Terraform (QA + Prod) adds `ows_account_cache` ElastiCache module, `datadog_elasticache` monitoring module, and `REDIS_URL` / `REDIS_CACHE_TTL` Fargate environment variables.
- Manual smoke test: two consecutive calls return the same result; updating restricted features via the write endpoint causes the next lookup to reflect the change.

#### Implementation Details

**Terraform** — add to `qa/ows-account/main.tf` and `prod/ows-account/main.tf`:

```hcl
module "ows_account_cache" {
  source = "git@github.com:theorchard/terraform-elasticache.git//?ref=4.0.0"

  providers = { aws.dns = aws.networking }

  environment                    = var.environment
  service_name                   = var.service_name
  application_family             = var.application_family
  cache_engine                   = "redis"
  redis_engine_version           = "7.1"
  cache_subnet_group_name        = "${var.environment}-elasticache-subnet-group"
  cache_node_type                = "cache.t4g.micro"
  cache_parameter_group_name     = "default.redis7"
  vpc_id                         = module.vpc_info.vpc_id
  additional_cidr_blocks_enabled = "true"
  additional_cidr_blocks         = [...]
}

module "datadog_elasticache" {
  source = "git@github.com:theorchard/terraform-datadog.git//modules/elasticache?ref=6.16.1"

  environment        = var.environment
  service_name       = var.service_name
  application_family = var.application_family
  teams              = var.teams

  notification_endpoints            = var.application_alert_channel
  escalation_notification_endpoints = var.application_alert_channel
}
```

Add to `environment_variables` in the Fargate module:

```hcl
{ REDIS_URL      = module.ows_account_cache.redis_primary_endpoint_address },
{ REDIS_CACHE_TTL = var.REDIS_CACHE_TTL },
```

**New file: `account/connectors/redis.py`**

```python
"""Connector for Redis."""

import json

import redis
from owsresponse.adaptors.flask_encoder import FlaskEncoder

from account import config

if not config.REDIS_URL:
    import fakeredis
    client = fakeredis.FakeStrictRedis()
else:
    client = redis.Redis(host=config.REDIS_URL)


def set(key, data, ttl=config.REDIS_CACHE_TTL):  # noqa: A001
    return client.set(key, json.dumps(data, cls=FlaskEncoder), ex=ttl)


def get(key):
    data = client.get(key)
    if data is None:
        return None
    return json.loads(data.decode('utf8'))


def delete_keys(keys: list[str]) -> None:
    if keys:
        client.delete(*keys)
```

**Modified file: `account/config.py`**

```python
REDIS_URL = os.environ.get('REDIS_URL', None)
REDIS_CACHE_TTL = int(os.environ.get('REDIS_CACHE_TTL', 3600))
```

**Modified file: `account/handlers/lookups.py`** — cache-aside reads and writes

```python
from account.connectors import redis as redis_connector

VENDOR_FEATURES_CACHE_KEY = "vendor_features:{uuid}"
SUBACCOUNT_FEATURES_CACHE_KEY = "subaccount_features:{uuid}"


@app.route('/lookup/vendors/features/uuids/', methods=['POST'])
@validate_request_data(LookupVendorFeaturesByUuids())
def lookup_vendors_features_by_uuids(deserialize_schema):
    """NOTE: No access rule checks — PIP endpoint for Permission Platform."""
    uuids = [str(u) for u in deserialize_schema['uuids']]

    cache_hits, missing = {}, []
    for uuid in uuids:
        cached = redis_connector.get(VENDOR_FEATURES_CACHE_KEY.format(uuid=uuid))
        if cached is not None:
            cache_hits[uuid] = cached
        else:
            missing.append(uuid)

    if missing:
        result = feature.lookup_features_by_vendor_uuids(missing)
        for entry in result.message['vendors']:
            redis_connector.set(
                VENDOR_FEATURES_CACHE_KEY.format(uuid=entry['uuid']),
                entry['feature_ids'],
            )
            cache_hits[entry['uuid']] = entry['feature_ids']

    vendors = [{'uuid': u, 'feature_ids': cache_hits.get(u, [])} for u in uuids]
    return flaskify(response.Response({'vendors': vendors}))


# lookup_subaccount_features_by_uuids follows the same pattern,
# using SUBACCOUNT_FEATURES_CACHE_KEY and feature.lookup_features_by_subaccount_uuids.
```

**Modified file: `account/handlers/vendor.py`** — cache invalidation on write

```python
from account.connectors import redis as redis_connector
from account.handlers.lookups import VENDOR_FEATURES_CACHE_KEY, SUBACCOUNT_FEATURES_CACHE_KEY


def _feature_cache_keys_for_vendor_id(vendor_id, session):
    """Return cache keys for the vendor and all its subaccounts."""
    vendor_uuid = feature_model.get_vendor_uuid_by_id(vendor_id, session)
    subaccount_uuids = feature_model.get_subaccount_uuids_by_vendor_id(vendor_id, session)
    keys = [VENDOR_FEATURES_CACHE_KEY.format(uuid=vendor_uuid)]
    keys += [SUBACCOUNT_FEATURES_CACHE_KEY.format(uuid=u) for u in subaccount_uuids]
    return keys


# In bulk_add_restricted_features_for_vendor, after the successful write:
try:
    redis_connector.delete_keys(_feature_cache_keys_for_vendor_id(vendor_id, session))
except Exception as e:
    g.log.warning(f'Failed to bust vendor features cache for vendor {vendor_id}: {e}')

# Apply the same invalidation block in bulk_remove_restricted_features_for_vendor.
```

#### Notes

- `get_vendor_uuid_by_id` and `get_subaccount_uuids_by_vendor_id` may already exist in `account/models/`. Add them to `account/models/feature.py` or `account/models/vendor.py` if not — they are simple indexed lookups and do not warrant their own ticket.
- Cache TTL is a safety-net backstop; explicit invalidation on write is the primary consistency mechanism. The TTL default of 3600 s is intentionally generous since feature controls change infrequently.
- `fakeredis` fallback means local development and unit tests require no running Redis instance, matching the ows-permissions pattern.
- The `additional_cidr_blocks` list in Terraform should match the blocks used in the ows-permissions and ows-pdp ElastiCache modules for the same environment. Confirm with the infra team before merging.