bdd plan

gistfile1.txt

# Multi-Service Flow Orchestration Pattern

**Date:** 2026-03-16
**Status:** Approved

## Overview

Design for a **reusable multi-service API orchestration pattern** (the "Flow" layer) within the existing BDD test framework. A flow chains multiple services where each stage's output feeds into the next.

The Data Product Registration flow below is the **first implementation** of this pattern, serving as the reference for all future multi-service flows.

### Generic Flow Pattern

```
Flow Orchestrator
  ├── ServiceA.operation() → extract value
  ├── ServiceB.operation(value_from_A) → extract next value
  └── ServiceC.operation(value_from_B) → (single or batch)
```

Any multi-service chain that crosses service boundaries should use this pattern rather than encoding the logic in step definitions or fixtures.

### Example Flow: Data Product Registration

```
Platform API (get options → match by name → extract ID)
  → Data Product API (register with platform option ID → get data_product_id)
    → Schema API (register N schemas against data_product_id)
```

### Constraints

- Services within a flow share the same base URL (different path prefixes) and the same auth context
- **Each service in a flow may target a different API version** (e.g., Platform v3, Data Product v4, Schema v4) — versions are parameterized from the BDD layer
- Every flow must work as both a **test target** (validating the flow itself) and a **reusable setup** (precondition for downstream tests)
- Ordering within batch operations is scenario-dependent — driven by BDD parameterization
- Failures fail the test; no resume, rollback, or state machine needed

## Architecture

### Layer Integration

New `flows/` layer sits between steps and services:

```
Feature (.feature)        ← parameterized scenarios
  → Step Definitions      ← thin glue
    → Flows               ← NEW: multi-service orchestration
      → Services          ← single-resource CRUD
        → Client          ← SyncAPIClient (shared base URL + auth)
```

**Flow Layer Rules (apply to ALL flows, not just this example):**
1. Flows depend on **Services** (constructor injection), never touch the HTTP client directly
2. Steps can call a **Flow** (end-to-end or setup) or individual **Services** (single-resource tests)
3. Flows do **NOT assert** — they return a result dataclass; assertions belong in steps/tests
4. Every flow has a **result dataclass** that preserves all intermediate `APIResponse` objects
5. Every public method and stage is decorated with `@allure.step()` for reporting visibility
6. Flows are **stateless** — call `execute()` and get a result. No side effects beyond API calls
7. Payloads are **never mutated** — use `model_copy(update={...})` for immutable updates
8. Flow files live in `flows/` directory, one file per flow: `flows/{flow_name}.py`
9. Each flow defines a **`FlowVersions` dataclass** with per-service version fields (all default to `None`). The `execute()` method accepts an optional `versions` param so BDD scenarios can drive per-service API version combinations

### When to Use a Flow vs. a Service

| Scenario | Use |
|----------|-----|
| Single API call (CRUD on one resource) | Service directly |
| Multiple API calls on the **same** resource (e.g., create then get) | Service directly |
| Multiple API calls across **different** services where output feeds into input | Flow |
| Reusable setup that involves multiple services | Flow |

## Per-Service API Versioning

### Problem

The current framework sets `api_version` once on the `BaseAPIClient` at init time (line 82 of `base_client.py`), which bakes a single version into every URL via `_prepare_request()`. In a multi-service flow, each service may target a different API version (e.g., Platform v3, Data Product v4, Schema v4).

### Solution: Per-Request Version Override

Add an optional `api_version` override to `BaseService._request()` and to the underlying `BaseAPIClient.request()`. When provided, it overrides the client-level default for that single request.

**`core/client/base_client.py` — changes to `_prepare_request()`:**

```python
def _prepare_request(
    self, method: str, endpoint: str, **kwargs: Any
) -> PreparedRequest:
    # Per-request version override takes precedence over client default
    version = kwargs.pop("api_version", self.api_version)
    if version is not None:
        url = f"{self.base_url}/api/{version}/{endpoint.lstrip('/')}"
    else:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
    # ... rest unchanged
```

**`services/base_service.py` — changes to `_request()`:**

```python
def _request(
    self,
    method: str,
    endpoint: str,
    *,
    body: BaseModel | None = None,
    params: dict[str, Any] | None = None,
    api_version: str | None = None,   # NEW: per-request version override
    **kwargs: Any,
) -> APIResponse:
    if body is not None:
        kwargs["json"] = body.model_dump(by_alias=True, exclude_none=True)
    if params is not None:
        kwargs["params"] = params
    if api_version is not None:
        kwargs["api_version"] = api_version
    return self.client.request(method, endpoint, **kwargs)
```

**Each service method accepts an optional `api_version`:**

```python
class DataProductService(BaseService):
    @allure.step("Register data product")
    def register(self, payload: CreateDataProductRequest, *, api_version: str | None = None) -> APIResponse:
        return self._request("POST", Endpoints.DATA_PRODUCTS.url(), body=payload, api_version=api_version)
```

### Backward Compatibility

- When `api_version` is not passed to a service method → not passed to `_request()` → not in `kwargs` → `kwargs.pop("api_version", self.api_version)` falls back to the client default
- All existing code continues to work unchanged — single-version tests are unaffected
- The override is opt-in at every layer

### Flow Integration

Flows accept a version map and pass per-service versions through:

```python
@dataclass
class FlowVersions:
    """Per-service API version overrides for a flow."""
    platform: str | None = None
    data_product: str | None = None
    schema: str | None = None
```

```python
class DataProductRegistrationFlow:
    @allure.step("Execute data product registration flow")
    def execute(
        self,
        option_name,
        product_payload,
        schema_payloads,
        versions: FlowVersions | None = None,
    ) -> RegistrationResult:
        v = versions or FlowVersions()

        # Step 1: Resolve platform option (using platform version)
        options_response = self.platform_service.get_options(api_version=v.platform)
        option = self.platform_service.find_option_by_name(options_response, option_name)

        # Step 2: Register data product (using data_product version)
        updated_payload = product_payload.model_copy(update={"platform_option_id": option["id"]})
        product_response = self.data_product_service.register(updated_payload, api_version=v.data_product)
        data_product_id = product_response.json_data["id"]

        # Step 3: Register schemas (using schema version)
        schema_responses = self.schema_service.register_batch(
            data_product_id, schema_payloads, api_version=v.schema
        )

        return RegistrationResult(...)
```

### BDD Parameterization

Versions are driven from the feature file via Scenario Outline:

```gherkin
@regression
Scenario Outline: Register data product across API versions
  Given platform option "<option_name>" is available on version "<platform_version>"
  When I register a data product with platform option "<option_name>" on version "<dp_version>"
  And I register schemas "<schemas>" for the data product on version "<schema_version>"
  Then <schema_count> schemas should be registered

  Examples:
    | option_name | platform_version | dp_version | schema_version | schemas   | schema_count |
    | raw_events  | v3               | v4         | v4             | avro,json | 2            |
    | analytics   | v3               | v3         | v4             | parquet   | 1            |
```

Step definitions parse the version and pass it through:

```python
@when(parsers.parse('I register a data product with platform option "{option_name}" on version "{dp_version}"'), target_fixture="api_response")
def register_dp_versioned(data_product_service, platform_option, data_product_factory, dp_version):
    payload = data_product_factory.create()
    updated = payload.model_copy(update={"platform_option_id": platform_option["id"]})
    return data_product_service.register(updated, api_version=dp_version)
```

Or when using the full flow:

```python
@when('I execute the registration flow', target_fixture="registration_result")
def execute_flow(registration_flow, option_name, product_payload, schema_payloads, flow_versions):
    return registration_flow.execute(option_name, product_payload, schema_payloads, versions=flow_versions)
```

Where `flow_versions` is built from BDD parameters in a preceding step:

```python
@given(parsers.parse('service versions platform="{pv}" data_product="{dpv}" schema="{sv}"'), target_fixture="flow_versions")
def set_versions(pv, dpv, sv):
    return FlowVersions(platform=pv, data_product=dpv, schema=sv)
```

### Generic Pattern for Future Flows

Every `FlowVersions` dataclass is flow-specific (different flows have different services). The convention:

- File: `flows/{flow_name}.py` — contains both the flow class and its `FlowVersions` dataclass
- Field names match service names in the flow
- All fields default to `None` (fall back to client default)

## New Services

All follow the existing `BaseService` pattern.

### PlatformService (`services/platform_service.py`)

- `get_options() -> APIResponse` — fetches available platform options
- `@staticmethod find_option_by_name(options_response, name: str) -> dict` — pure logic helper, extracts matching option from response. Static method since it performs no HTTP call — keeps services as HTTP-resource wrappers consistent with `UserService`/`AuthService`.

### DataProductService (`services/data_product_service.py`)

- `register(payload: CreateDataProductRequest) -> APIResponse`
- `get(data_product_id: str) -> APIResponse`
- `list() -> APIResponse`

### SchemaService (`services/schema_service.py`)

- `register(data_product_id: str, payload: CreateSchemaRequest) -> APIResponse`
- `register_batch(data_product_id: str, payloads: list[CreateSchemaRequest]) -> list[APIResponse]` — sequential registration; on failure, returns partial results collected so far then raises, enabling Allure to report which schemas succeeded before the failure
- `get(data_product_id: str, schema_id: str) -> APIResponse`

## Flow Orchestrator

### DataProductRegistrationFlow (`flows/data_product_registration.py`)

```python
class DataProductRegistrationFlow:
    def __init__(self, platform_service, data_product_service, schema_service):
        ...

    @allure.step("Execute data product registration flow")
    def execute(self, option_name, product_payload, schema_payloads) -> RegistrationResult:
        # Step 1: Resolve platform option
        options_response = self._resolve_platform_option(option_name)
        option = self.platform_service.find_option_by_name(options_response, option_name)

        # Step 2: Register data product (immutable payload update)
        updated_payload = product_payload.model_copy(update={"platform_option_id": option["id"]})
        product_response = self._register_data_product(updated_payload)
        data_product_id = product_response.json_data["id"]

        # Step 3: Register schemas
        schema_responses = self._register_schemas(data_product_id, schema_payloads)

        return RegistrationResult(
            platform_option=option,
            data_product_response=product_response,
            schema_responses=schema_responses,
            data_product_id=data_product_id,
            schema_ids=[r.json_data["id"] for r in schema_responses],
        )

    @allure.step("Resolve platform option: {option_name}")
    def _resolve_platform_option(self, option_name):
        return self.platform_service.get_options()

    @allure.step("Register data product")
    def _register_data_product(self, payload):
        return self.data_product_service.register(payload)

    @allure.step("Register schemas for data product: {data_product_id}")
    def _register_schemas(self, data_product_id, schema_payloads):
        return self.schema_service.register_batch(data_product_id, schema_payloads)
```

**Design decisions:**
- Every intermediate `APIResponse` is preserved in the result — nothing swallowed
- Constructor injection of all three services (testable, mockable)
- `@allure.step()` decorators on each stage for reporting visibility
- Stateless — call `execute()` and get a result

### RegistrationResult (`flows/data_product_registration.py`)

```python
@dataclass
class RegistrationResult:
    platform_option: dict
    data_product_response: APIResponse
    schema_responses: list[APIResponse]
    data_product_id: str
    schema_ids: list[str]  # extracted convenience field
```

## Models & Data

### Pydantic DTOs

**`models/data_product.py`**
- `CreateDataProductRequest` — `name`, `description`, `platform_option_id`, `product_type`
- `UpdateDataProductRequest` — all fields optional

**`models/schema.py`**
- `CreateSchemaRequest` — `name`, `format` (avro/json/parquet), `definition` (schema body), `version`

### Builders

- `DataProductBuilder` — `.with_name()`, `.with_type()`, `.with_platform_option_id()`, `.with_defaults()`
- `SchemaBuilder` — `.with_name()`, `.with_format()`, `.with_definition()`, `.with_defaults()`

### Factories

- `DataProductFactory` — `create()`, `streaming()`, `batch()`, `create_batch(n)`
- `SchemaFactory` — `avro()`, `json_schema()`, `parquet()`, `create_batch(formats: list[str])`

## Endpoints

New members added to the existing `Endpoints(str, Enum)` in `config/endpoints.py`:

```python
class Endpoints(str, Enum):
    # ... existing members ...
    PLATFORM_OPTIONS = "/platform/options"
    DATA_PRODUCTS = "/data-products"
    DATA_PRODUCT_BY_ID = "/data-products/{data_product_id}"
    DATA_PRODUCT_SCHEMAS = "/data-products/{data_product_id}/schemas"
    DATA_PRODUCT_SCHEMA_BY_ID = "/data-products/{data_product_id}/schemas/{schema_id}"
```

Services use the `.url()` method for path parameter rendering:
```python
Endpoints.DATA_PRODUCT_BY_ID.url(data_product_id="abc-123")
# => "/data-products/abc-123"
Endpoints.DATA_PRODUCT_SCHEMAS.url(data_product_id="abc-123")
# => "/data-products/abc-123/schemas"
```

## Generated Response Models

Added to OpenAPI specs and generated via `datamodel-code-generator`:

- `PlatformOptionsResponse` — list of options with `id`, `name`, `type`
- `DataProductResponse` — `id`, `name`, `platform_option_id`, `status`
- `DataProductListResponse` — paginated list
- `SchemaResponse` — `id`, `data_product_id`, `name`, `format`, `version`
- `SchemaListResponse` — list of schemas

Model registry resolves these via `resolve_model("data_product_response", version="v1")`.

## Fixtures

New file: `fixtures/conftest_data_product.py`

```python
# Session-scoped services (consistent with existing user_service pattern in conftest_data.py)
@pytest.fixture(scope="session")
def platform_service(authenticated_client) -> PlatformService

@pytest.fixture(scope="session")
def data_product_service(authenticated_client) -> DataProductService

@pytest.fixture(scope="session")
def schema_service(authenticated_client) -> SchemaService

# Function-scoped factories (new data per test)
@pytest.fixture
def data_product_factory() -> DataProductFactory

@pytest.fixture
def schema_factory() -> SchemaFactory

# Session-scoped flow (reuses session-scoped services)
@pytest.fixture(scope="session")
def registration_flow(platform_service, data_product_service, schema_service) -> DataProductRegistrationFlow

# Session-scoped setup fixture for use as precondition in downstream tests
@pytest.fixture(scope="session")
def registered_data_product(registration_flow, data_product_factory, schema_factory) -> RegistrationResult
```

Note: Service and flow fixtures are session-scoped (matching existing `user_service` pattern) to avoid scope mismatch with `registered_data_product`. Factories remain function-scoped for fresh data per test.

## BDD Integration

### Feature File (`features/data_product/register_data_product.feature`)

```gherkin
@data_product @regression
Feature: Data Product Registration

  Background:
    Given the API is available

  @smoke @critical
  Scenario: Full registration flow
    Given platform option "raw_events" is available
    When I register a data product with platform option "raw_events"
    And I register schemas "avro,json" for the data product
    Then the response status code should be 201
    And 2 schemas should be registered

  @regression
  Scenario Outline: Register with different configurations
    Given platform option "<option_name>" is available
    When I register a data product of type "<product_type>" with platform option "<option_name>"
    And I register schemas "<schemas>" for the data product
    Then <schema_count> schemas should be registered

    Examples:
      | option_name | product_type | schemas       | schema_count |
      | raw_events  | streaming    | avro,json     | 2            |
      | analytics   | batch        | parquet       | 1            |
      | ml_features | streaming    | avro          | 1            |
```

### Step Definitions (`steps/data_product_steps.py`)

Steps use `target_fixture` to pass data between steps (consistent with existing `auth_steps.py` and `user_steps.py` patterns):

```python
# Setup steps
@given(parsers.parse('platform option "{option_name}" is available'), target_fixture="platform_option")
# Calls platform_service.get_options() + find_option_by_name(), stores option dict

@given(parsers.parse('a data product of type "{product_type}" is registered'), target_fixture="registration_result")
# Calls registration_flow.execute(), stores RegistrationResult

# Action steps
@when(parsers.parse('I register a data product with platform option "{option_name}"'), target_fixture="api_response")
# Calls data_product_service.register() using platform_option from previous step, stores APIResponse

@when(parsers.parse('I register a data product of type "{product_type}" with platform option "{option_name}"'), target_fixture="api_response")
# Builds payload with product_type, calls data_product_service.register(), stores APIResponse

@when(parsers.parse('I register schemas "{schema_list}" for the data product'), target_fixture="schema_responses")
# Parses comma-separated schema_list, calls schema_service.register_batch(), stores list[APIResponse]

# Assertion steps
@then("the data product should have id")
# Asserts api_response.json_data["id"] is not None

@then(parsers.parse("{count:d} schemas should be registered"))
# Asserts len(schema_responses) == count
```

Inter-step data flow: `target_fixture` injects results into the pytest request scope so subsequent steps can access them by fixture name (e.g., `platform_option`, `api_response`, `schema_responses`).

Existing `common_steps.py` assertions (status code, schema validation, response time) apply without changes — they read from `api_response` which is set by the action steps above.

## .github Updates

The `.github/` instructions and prompts must be updated to reflect the new `flows/` layer so that all AI-assisted code generation follows the correct architecture.

### 1. `copilot-instructions.md` — Updates

**Call Chain** — update to include the flows layer:

```
Feature (.feature) -> Step (steps/) -> Flow (flows/) or Service (services/) -> Client (core/client/) -> HTTP/Messaging
```

**Design Patterns table** — add new row:

| Pattern | Location | Key Classes |
|---------|----------|-------------|
| Flow Orchestrator | `flows/` | `DataProductRegistrationFlow` — chains services, returns result dataclass |

**Architecture Rules** — add new rules:

> 7. **Multi-service chains go through Flows** — when a test requires calling 2+ services where output feeds into input, create a flow in `flows/`. Flows depend on services (constructor injection), never touch the HTTP client, and never assert. They return a result dataclass preserving all intermediate responses.
> 8. **Per-service versioning in flows** — each service method accepts an optional `api_version` parameter. Flows accept a `FlowVersions` dataclass to route each service call to its own API version. BDD Scenario Outlines drive version combinations.

**Service Object row** — update to include new services:

```
| Service Object | `services/` | `BaseService`, `UserService`, `AuthService`, `PlatformService`, `DataProductService`, `SchemaService`, `EventHubService`, `KafkaService` |
```

**Key DTOs table** — add new row:

| Dataclass | Location | Purpose |
|-----------|----------|---------|
| `RegistrationResult` | `flows/data_product_registration.py` | Multi-service flow result (all intermediate responses + extracted IDs) |

**Fixture Scoping table** — add new entries:

| Scope | Fixtures |
|-------|----------|
| session | (add) `platform_service`, `data_product_service`, `schema_service`, `registration_flow` |
| function | (add) `data_product_factory`, `schema_factory` |

**File Conventions** — add new line:

```
- `flows/{flow_name}.py` — multi-service orchestration flows (one flow per file)
```

**Adding a New API Domain** — add step 10 (renumber existing 9 to 10):

> 10. If the domain participates in a multi-service flow, create or update a flow in `flows/`

### 2. `prompts/add-api-domain.prompt.md` — Updates

**Call Chain** — update to include flows:

```
Feature (.feature) -> Step (steps/) -> Flow (flows/) or Service (services/) -> Client (core/client/) -> HTTP
```

**Add new section after step 11 (Markers):**

```markdown
### 12. Multi-Service Flows (if applicable) — `flows/{flow_name}.py`
If this domain participates in a chain with other services (e.g., Service A output feeds Service B input), create or update a flow orchestrator:
```python
class MyFlow:
    def __init__(self, service_a: ServiceA, service_b: ServiceB):
        self._a = service_a
        self._b = service_b

    @allure.step("Execute my flow")
    def execute(self, ...) -> MyFlowResult:
        resp_a = self._a.operation(...)
        value = resp_a.json_data["id"]
        resp_b = self._b.operation(value, ...)
        return MyFlowResult(a_response=resp_a, b_response=resp_b, ...)
```

Flow rules:
- Constructor injection of all services
- Never touch HTTP client directly
- Never assert — return a result dataclass
- Every stage decorated with `@allure.step()`
- Payloads never mutated — use `model_copy(update={...})`
- Add a session-scoped fixture in `fixtures/conftest_{domain}.py`
```

**Rules section** — add:

```
- Multi-service chains MUST use a Flow orchestrator — never chain service calls in step definitions
```

**Checklist** — add:

```
- [ ] Flow orchestrator in `flows/{flow_name}.py` (if multi-service)
- [ ] Flow fixture in `fixtures/conftest_{domain}.py` (if multi-service)
```

### 3. `prompts/add-scenario.prompt.md` — Updates

**Rules** — add new rule:

```
11. If the scenario involves calling 2+ services where output feeds input, delegate to a Flow (`flows/`) — never chain service calls directly in step definitions
```

**Add to the "Reuse shared steps" list:**

```
- Given steps that use `target_fixture="registration_result"` (or similar flow results) for multi-service setup
```

### 4. `prompts/modify-scenario.prompt.md` — Updates

**Add to pre-modification checks:**

```
- Check if the scenario uses a Flow (`flows/`). If it does, modifications to the multi-service chain must be made in the flow, not in the step definitions.
```

### 5. `prompts/remove-scenario.prompt.md` — Updates

**Add to orphaned artifact cleanup:**

```
- **Flow cleanup**: If the removed scenario was the only consumer of a flow, check if the flow is still referenced elsewhere. If not, remove the flow file from `flows/` and its fixture from `fixtures/`.
```

### 6. New Prompt: `prompts/add-multi-service-flow.prompt.md`

A new prompt file for scaffolding multi-service flows:

```markdown
# Add a Multi-Service Flow

Scaffold a new flow orchestrator that chains multiple services.

## Input Required
- Flow name (e.g., `data_product_registration`)
- Services involved and their order
- What data passes between stages (field names, extraction logic)
- Whether it will be used as test target, setup, or both

## Call Chain (never skip a layer)
```
Feature (.feature) -> Step (steps/) -> Flow (flows/) -> Service (services/) -> Client (core/client/) -> HTTP
```

## Files to Create (in order)

### 1. Services (if new) — follow `add-api-domain.prompt.md` steps 1-6 for each

### 2. Result Dataclass — `flows/{flow_name}.py`
```python
@dataclass
class MyFlowResult:
    # Every intermediate APIResponse preserved
    step_a_response: APIResponse
    step_b_response: APIResponse
    # Convenience extracted IDs
    resource_id: str
```

### 3. Flow Orchestrator — `flows/{flow_name}.py`
```python
class MyFlow:
    def __init__(self, service_a: ServiceA, service_b: ServiceB):
        self._a = service_a
        self._b = service_b

    @allure.step("Execute my flow")
    def execute(self, ...) -> MyFlowResult:
        resp_a = self._resolve_step_a(...)
        value = resp_a.json_data["id"]
        resp_b = self._execute_step_b(value, ...)
        return MyFlowResult(
            step_a_response=resp_a,
            step_b_response=resp_b,
            resource_id=resp_b.json_data["id"],
        )

    @allure.step("Step A: {description}")
    def _resolve_step_a(self, ...):
        return self._a.operation(...)

    @allure.step("Step B: {description}")
    def _execute_step_b(self, value, ...):
        return self._b.operation(value, ...)
```

### 4. Fixtures — `fixtures/conftest_{domain}.py`
```python
@pytest.fixture(scope="session")
def my_flow(service_a, service_b) -> MyFlow:
    return MyFlow(service_a, service_b)
```

### 5. Steps — `steps/{domain}_steps.py`
- Setup steps store flow results via `target_fixture="flow_result"`
- Action steps call individual services or the full flow
- Assertion steps read from `flow_result` or `api_response`

### 6. Features — `features/{domain}/*.feature`
- Use Scenario Outline for parameterized variations
- Each scenario must be independently runnable

## Per-Service Versioning
Each service method accepts an optional `api_version` param. Flows accept a `FlowVersions` dataclass:
```python
@dataclass
class MyFlowVersions:
    service_a: str | None = None
    service_b: str | None = None
```
Pass versions from BDD Scenario Outline:
```gherkin
Given service versions service_a="v3" service_b="v4"
```

## Rules
- Flows depend on Services only — never touch HTTP client
- Flows NEVER assert — return result dataclass
- Every stage decorated with `@allure.step()`
- Payloads never mutated — use `model_copy(update={...})`
- Constructor injection for all dependencies
- Step defs max 5 lines — delegate to flow or service
- Result dataclass preserves ALL intermediate APIResponse objects
- Each service call passes `api_version` from `FlowVersions` — never hardcode versions in flows

## Checklist
- [ ] All participating services created/exist (follow add-api-domain if new)
- [ ] `FlowVersions` dataclass in `flows/{flow_name}.py`
- [ ] Result dataclass in `flows/{flow_name}.py`
- [ ] Flow orchestrator in `flows/{flow_name}.py`
- [ ] `flows/__init__.py` exports flow, result, and versions classes
- [ ] Session-scoped flow fixture in `fixtures/conftest_{domain}.py`
- [ ] Step definitions in `steps/{domain}_steps.py` (including version-passing steps)
- [ ] Feature file in `features/{domain}/` with version-parameterized Scenario Outlines
- [ ] Test collector in `tests/{domain}/`
- [ ] `conftest.py` imports new fixture module
```

## Out of Scope (Planned for Later)

- **Negative test scenarios:** Platform option not found, duplicate data product (409), invalid schema format, unauthorized access. These will be added as separate feature files per service.
- **DataProductService.delete():** Needed for test cleanup/teardown. Will be added when cleanup fixtures are implemented.
- **Async flow execution:** Current flow uses sync client only. Async variant can be added if performance requires it.

## File Summary

### New Files

| File | Purpose |
|------|---------|
| `flows/__init__.py` | Flow module init — exports `DataProductRegistrationFlow`, `RegistrationResult` |
| `flows/data_product_registration.py` | Orchestrator + RegistrationResult |
| `services/platform_service.py` | Platform options CRUD |
| `services/data_product_service.py` | Data product CRUD |
| `services/schema_service.py` | Schema CRUD |
| `models/data_product.py` | CreateDataProductRequest, UpdateDataProductRequest |
| `models/schema.py` | CreateSchemaRequest |
| `models/builders/data_product_builder.py` | Fluent builder |
| `models/builders/schema_builder.py` | Fluent builder |
| `data/factories/data_product_factory.py` | Test data factory |
| `data/factories/schema_factory.py` | Test data factory |
| `fixtures/conftest_data_product.py` | pytest fixtures |
| `steps/data_product_steps.py` | BDD step definitions |
| `features/data_product/register_data_product.feature` | Feature file |

### New Files

| File | Purpose |
|------|---------|
| `.github/prompts/add-multi-service-flow.prompt.md` | Prompt for scaffolding new multi-service flows |

### Modified Files

| File | Change |
|------|--------|
| `core/client/base_client.py` | Add per-request `api_version` override in `_prepare_request()` |
| `services/base_service.py` | Add optional `api_version` param to `_request()` |
| `config/endpoints.py` | Add 5 new endpoint entries |
| `conftest.py` | Import new fixture module |
| `openapi/v1.yaml` | Add Platform, DataProduct, Schema endpoints + schemas |
| `models/generated/` | Regenerated after OpenAPI update |
| `.github/copilot-instructions.md` | Add flows layer, per-service versioning, patterns, rules, conventions |
| `.github/prompts/add-api-domain.prompt.md` | Add flow step, update call chain, add checklist items |
| `.github/prompts/add-scenario.prompt.md` | Add flow delegation rule |
| `.github/prompts/modify-scenario.prompt.md` | Add flow awareness to pre-modification checks |
| `.github/prompts/remove-scenario.prompt.md` | Add flow cleanup to orphan detection |