# Phase 7 — Composite Aggregation (Fan-out, Priority & Fallback)

## Overview

Enable MW-Core to handle incoming requests that require fetching data from **multiple
heterogeneous backends in parallel** (legacy systems, REST APIs, direct database), then
**merging the results into a single unified response** before returning to the caller.

A critical requirement within this pattern is **source priority with fallback chains**:
when a primary source (e.g. CRM API) is unavailable or returns no usable data, the system
must automatically fall back to a secondary source (e.g. local database) for that data
group — transparently, without the caller needing to know or retry.

---

## Problem Statement

Today `MwRouter.Dispatcher` dispatches to **exactly one adapter** per request. There is no
mechanism to:

- Fire requests to multiple adapters concurrently
- Define priority chains: primary source → fallback source(s) per data group
- Detect "empty" responses (HTTP 200 but no usable data) and trigger a fallback
- Merge N adapter responses into one normalized payload
- Surface per-source outcome metadata (used / fallback-used / unavailable) to callers

This phase adds that capability as a first-class, config-driven feature — without breaking
any existing single-adapter routes.

---

## Core Concept: Slots + Priority Chains

The fundamental modelling unit is a **Slot** — a named logical data group within the
composite response. Each slot has an **ordered chain** of adapters: the first adapter in
the chain is the primary source; subsequent adapters are fallbacks, tried in order.

```
Slot "customer_profile"
  ├── priority 1 → crm_api          (primary)
  └── priority 2 → local_db         (fallback if crm_api fails or returns empty)

Slot "account_balance"
  └── priority 1 → banking_legacy   (required, no fallback)

Slot "transaction_history"
  ├── priority 1 → banking_legacy   (primary)
  └── priority 2 → adapter_dw       (fallback — read from data warehouse)
```

### Fallback Trigger Conditions

A fallback is triggered for a slot when the current highest-priority adapter in the chain
produces any of the following:

| Trigger | Meaning |
|---------|---------|
| `:timeout` | Adapter exceeded its configured `timeout_ms` |
| `:error` | Adapter returned an error (HTTP 4xx/5xx, connection failure, exception) |
| `:circuit_open` | Fuse circuit breaker is blown for this adapter |
| `:empty` | Adapter returned HTTP 200 but body is nil, `%{}`, or `[]` — *current data not available* |

The `:empty` trigger is essential: CRM returning HTTP 200 with no records is functionally
equivalent to a timeout from the caller's perspective, and the DB fallback should activate.

### Execution Strategy: Parallel-Fire, Priority-Select

All adapters within a slot fire **concurrently**. After results are collected, the slot
resolves to the highest-priority result that passes its fallback condition check.

```
Slot "customer_profile" — fires both in parallel:
  Task A → crm_api   → {:ok, %{}}          ← empty!  fallback triggers
  Task B → local_db  → {:ok, %{name: ...}} ← valid   ✓ selected

  Resolution: local_db result used, meta records fallback_used: true
```

This avoids sequential latency — the DB query runs in parallel with CRM, so if CRM
empties, the DB result is already available with zero extra wait.

---

## Architecture

```
External Channel
      │
      ▼
 GatewayApi / GatewayMobile
      │
      ▼
 MwRouter.Pipeline  (unchanged — auth, rate limit, idempotency still apply)
      │
      ▼  route.type == :composite?
 MwRouter.FanoutDispatcher          ← NEW
      │
      ├── Slot: "customer_profile"
      │     ├── Task.async ──► crm_api     (priority 1)
      │     └── Task.async ──► local_db    (priority 2 / fallback)
      │
      ├── Slot: "account_balance"
      │     └── Task.async ──► banking_legacy (priority 1, required)
      │
      └── Slot: "transaction_history"
            ├── Task.async ──► banking_legacy (priority 1)
            └── Task.async ──► adapter_dw     (priority 2 / fallback)
                    │
                    ▼  collect all slot results, apply priority + fallback selection
      MwRouter.SlotResolver               ← NEW
      (per-slot: pick highest-priority passing result)
                    │
                    ▼
      MwTransform.Merger                  ← NEW
      (merge one result per slot into unified payload)
                    │
                    ▼
      Unified response  →  client
      {
        data: { customer_profile: {...}, account_balance: {...} },
        meta: {
          sources: {
            customer_profile: { used: "local_db", reason: "crm_api_empty" },
            account_balance:  { used: "banking_legacy" },
            transaction_history: { used: "banking_legacy" }
          }
        }
      }
```

---

## Sub-Phases

---

### Phase 7.0 — Schema & Contract Changes

**Goal:** Define the slot-based composite route model and extend kernel context to carry
per-slot results. Existing single-adapter routes are completely unaffected.

---

#### `mw_kernel` — new types

**`MwKernel.CompositeRoute`**

```elixir
# apps/mw_kernel/lib/mw_kernel/composite_route.ex

defmodule MwKernel.CompositeRoute do
  @moduledoc """
  Defines the structure of a composite (fan-out) route.
  A composite route contains one or more named Slots. Each Slot carries
  an ordered chain of ChainEntry adapters: priority 1 is the primary source,
  higher numbers are fallbacks tried in order.
  """

  defmodule ChainEntry do
    @moduledoc "One adapter within a slot's priority chain."
    @type fallback_trigger :: :timeout | :error | :circuit_open | :empty

    @type t :: %__MODULE__{
      adapter:      atom(),
      priority:     pos_integer(),
      timeout_ms:   pos_integer(),
      fallback_on:  [fallback_trigger()],
      field_map:    map() | nil
    }

    defstruct [
      :adapter,
      priority:    1,
      timeout_ms:  5_000,
      fallback_on: [:timeout, :error, :circuit_open, :empty],
      field_map:   nil
    ]
  end

  defmodule Slot do
    @moduledoc "A named logical data group with an ordered chain of adapter sources."
    @type t :: %__MODULE__{
      name:     atom(),
      required: boolean(),
      chain:    [ChainEntry.t()]
    }

    defstruct [:name, required: true, chain: []]
  end

  @type merge_strategy :: :deep_merge | :field_priority | :first_wins

  @type t :: %__MODULE__{
    slots:          [Slot.t()],
    merge_strategy: merge_strategy()
  }

  defstruct [slots: [], merge_strategy: :deep_merge]
end
```

**`MwKernel.Context`** — extend to carry slot results

```elixir
# apps/mw_kernel/lib/mw_kernel/context.ex  (additions only)
defstruct [
  ...existing fields...,
  slot_results: %{},    # %{slot_name => SlotResult.t()}
  source_meta:  %{}     # %{slot_name => %{used: adapter, reason: nil | string}}
]
```

**`MwKernel.SlotResult`** — result envelope per slot

```elixir
# apps/mw_kernel/lib/mw_kernel/slot_result.ex

defmodule MwKernel.SlotResult do
  @type outcome :: :ok | :fallback_used | :all_failed

  @type t :: %__MODULE__{
    slot:          atom(),
    outcome:       outcome(),
    data:          map() | nil,
    used_adapter:  atom() | nil,
    fallback_reason: atom() | nil,   # why the primary was skipped
    raw_results:   map()             # %{adapter => {:ok, resp} | {:error, reason}}
  }

  defstruct [:slot, :outcome, :data, :used_adapter, :fallback_reason, raw_results: %{}]
end
```

**Route table entry** — add `:composite` type

```elixir
%{
  path:           "/api/v1/query/customer-profile",
  method:         "GET",
  type:           :composite,
  adapter:        nil,
  composite_spec: %MwKernel.CompositeRoute{
    slots: [
      %Slot{
        name:     :customer_profile,
        required: false,
        chain: [
          %ChainEntry{adapter: :crm_api,  priority: 1, timeout_ms: 3_000,
                      fallback_on: [:timeout, :error, :circuit_open, :empty]},
          %ChainEntry{adapter: :local_db, priority: 2, timeout_ms: 1_000,
                      fallback_on: []}
        ]
      },
      %Slot{
        name:     :account_balance,
        required: true,
        chain: [
          %ChainEntry{adapter: :banking_legacy, priority: 1, timeout_ms: 5_000,
                      fallback_on: []}
        ]
      }
    ],
    merge_strategy: :deep_merge
  },
  tenant_id: "acme",
  active:    true
}
```

**Deliverable:** All types defined and compile. Existing `:single` routes are unaffected.
`RouteTable.init_tenant_table/1` and ETS serialisation accept the new shape.

**Tests:**
- `ChainEntry` default `fallback_on` includes all four trigger types
- `Slot` with empty chain raises compile-time NimbleOptions error
- `CompositeRoute` with no required slot still valid (all-optional composite)
- Route table ETS round-trip: `insert → lookup` preserves `composite_spec` struct

---

### Phase 7.1 — FanoutDispatcher + SlotResolver

**Goal:** For each slot in the composite spec, fire all chain entries in parallel, then
select the best result applying priority and fallback trigger logic.

**New modules:**
- `apps/mw_router/lib/mw_router/fanout_dispatcher.ex`
- `apps/mw_router/lib/mw_router/slot_resolver.ex`

---

#### `MwRouter.FanoutDispatcher`

Top-level orchestrator. Iterates over slots, fans out per-slot tasks in parallel across
slots (and within each slot), collects `SlotResult` per slot.

```elixir
defmodule MwRouter.FanoutDispatcher do
  alias MwKernel.{Context, CompositeRoute, SlotResult}
  alias MwRouter.SlotResolver
  require OpenTelemetry.Tracer, as: Tracer

  @spec dispatch(Context.t()) :: {:ok, Context.t()} | {:error, MwKernel.Error.t()}
  def dispatch(%Context{route: %{type: :composite, composite_spec: spec}} = ctx) do
    # Fan out: all slots fire in parallel. Within each slot, all chain entries fire
    # in parallel. Total wall-clock time ≈ max(slowest adapter across all slots).
    slot_results =
      spec.slots
      |> Task.async_stream(
           &resolve_slot(ctx, &1),
           timeout: :infinity,
           on_timeout: :kill_task,
           ordered: false
         )
      |> Enum.map(fn {:ok, result} -> {result.slot, result} end)
      |> Map.new()

    ctx = %{ctx |
      slot_results: slot_results,
      source_meta:  build_source_meta(slot_results)
    }

    case find_required_slot_failures(slot_results, spec.slots) do
      []       -> {:ok, ctx}
      failures -> {:error, MwKernel.Error.upstream_failure(failures)}
    end
  end

  # Resolve one slot: fire all chain entries in parallel, select winner
  defp resolve_slot(ctx, slot) do
    Tracer.with_span "fanout.slot.#{slot.name}" do
      raw_results =
        slot.chain
        |> Task.async_stream(
             &call_chain_entry(ctx, &1),
             timeout: :infinity,
             on_timeout: :kill_task,
             ordered: false
           )
        |> Enum.map(fn {:ok, r} -> r end)
        |> Map.new()

      SlotResolver.resolve(slot, raw_results)
    end
  end

  defp call_chain_entry(ctx, entry) do
    Tracer.with_span "fanout.adapter.#{entry.adapter}" do
      result =
        case Fuse.check(entry.adapter) do
          :ok    ->
            r = safe_call(ctx, entry)
            maybe_melt_fuse(entry.adapter, r)
            r
          :blown -> {:error, :circuit_open}
        end

      {entry.adapter, result}
    end
  end

  defp safe_call(ctx, %{adapter: name, timeout_ms: timeout}) do
    task = Task.async(fn -> MwRouter.Dispatcher.call_adapter(ctx, name) end)
    case Task.yield(task, timeout) || Task.shutdown(task) do
      {:ok, result} -> result
      nil           -> {:error, :timeout}
    end
  end

  defp build_source_meta(slot_results) do
    Map.new(slot_results, fn {slot_name, %SlotResult{} = r} ->
      {slot_name, %{used: r.used_adapter, fallback_reason: r.fallback_reason,
                    outcome: r.outcome}}
    end)
  end

  defp find_required_slot_failures(slot_results, slots) do
    slots
    |> Enum.filter(& &1.required)
    |> Enum.filter(fn slot ->
         match?(%SlotResult{outcome: :all_failed}, slot_results[slot.name])
       end)
    |> Enum.map(& &1.name)
  end

  defp maybe_melt_fuse(name, {:error, r}) when r != :circuit_open, do: Fuse.melt(name)
  defp maybe_melt_fuse(_, _), do: :ok
end
```

---

#### `MwRouter.SlotResolver`

Applies priority ordering and fallback trigger evaluation to pick the winning adapter
result for a slot.

```elixir
defmodule MwRouter.SlotResolver do
  @moduledoc """
  Given a slot's chain definition and the raw results from all chain entry calls,
  walks the chain in priority order and returns the first result that is NOT
  a fallback trigger condition.

  "Empty" detection: a response body of nil, %{}, or [] is treated as :empty.
  """

  alias MwKernel.{CompositeRoute.Slot, CompositeRoute.ChainEntry, SlotResult}

  @spec resolve(Slot.t(), map()) :: SlotResult.t()
  def resolve(%Slot{name: name, chain: chain}, raw_results) do
    chain
    |> Enum.sort_by(& &1.priority)
    |> Enum.reduce_while(:not_found, fn entry, _ ->
         result = raw_results[entry.adapter]
         case evaluate(result, entry.fallback_on) do
           {:use, data}           ->
             {:halt, build_result(name, :ok, data, entry.adapter, nil, raw_results)}

           {:fallback, reason}    ->
             {:cont, {:last_reason, reason, entry.adapter}}
         end
       end)
    |> case do
         %SlotResult{} = r -> r
         :not_found        -> build_result(name, :all_failed, nil, nil, nil, raw_results)
         {:last_reason, reason, _} ->
           build_result(name, :all_failed, nil, nil, reason, raw_results)
       end
  end

  # Evaluate whether a result is usable or triggers fallback
  defp evaluate({:ok, %{body: body}}, fallback_on) do
    cond do
      empty?(body) and :empty in fallback_on ->
        {:fallback, :empty}
      true ->
        {:use, body}
    end
  end

  defp evaluate({:error, :timeout}, fallback_on) when :timeout in fallback_on,
    do: {:fallback, :timeout}

  defp evaluate({:error, :circuit_open}, fallback_on) when :circuit_open in fallback_on,
    do: {:fallback, :circuit_open}

  defp evaluate({:error, _}, fallback_on) when :error in fallback_on,
    do: {:fallback, :error}

  defp evaluate({:ok, %{body: body}}, _fallback_on),
    do: {:use, body}

  defp evaluate(_, _),
    do: {:fallback, :error}

  defp empty?(nil),  do: true
  defp empty?(%{}),  do: true   # empty map
  defp empty?([]),   do: true   # empty list
  defp empty?(_),    do: false

  defp build_result(slot, outcome, data, adapter, reason, raw) do
    # Tag fallback usage in outcome
    final_outcome =
      if outcome == :ok and reason != nil, do: :fallback_used, else: outcome

    %SlotResult{
      slot:            slot,
      outcome:         final_outcome,
      data:            data,
      used_adapter:    adapter,
      fallback_reason: reason,
      raw_results:     raw
    }
  end
end
```

**Pipeline branch** — minimal change to `MwRouter.Pipeline`:

```elixir
# apps/mw_router/lib/mw_router/pipeline.ex
defp dispatch(ctx) do
  case ctx.route do
    %{type: :composite} -> MwRouter.FanoutDispatcher.dispatch(ctx)
    _                   -> MwRouter.Dispatcher.dispatch(ctx)     # unchanged
  end
end
```

**Deliverable:** Slot resolution with priority + fallback works end-to-end.

**Tests:**

| Scenario | Expected `SlotResult.outcome` | `used_adapter` |
|----------|-------------------------------|----------------|
| CRM returns data | `:ok` | `:crm_api` |
| CRM times out, DB returns data | `:fallback_used` | `:local_db` |
| CRM returns `{}` (empty), DB returns data | `:fallback_used` | `:local_db` |
| CRM circuit open, DB returns data | `:fallback_used` | `:local_db` |
| CRM times out, DB also times out | `:all_failed` | `nil` |
| Single-entry chain, adapter succeeds | `:ok` | adapter name |
| Empty body `[]` triggers `:empty` fallback | `:fallback_used` | fallback adapter |
| No `fallback_on` configured, CRM empty | `:ok` (empty is ok if not configured) | `:crm_api` |

---

### Phase 7.2 — Response Merger

**Goal:** Merge one resolved result per slot into a single unified response payload.
Apply per-slot field mappings before merging.

**New module:** `apps/mw_transform/lib/mw_transform/merger.ex`

```elixir
defmodule MwTransform.Merger do
  @moduledoc """
  Merges resolved SlotResults into a single unified response body.
  Each slot contributes its winning data (after field_map applied).
  Slots with outcome :all_failed contribute nothing to data but are
  reflected in meta.
  """

  alias MwKernel.{SlotResult, CompositeRoute.Slot}

  @type merged :: %{data: map(), meta: map()}

  @spec merge([SlotResult.t()], [Slot.t()], atom()) :: merged()
  def merge(slot_results, slots, strategy \\ :deep_merge) do
    {data_parts, degraded} =
      Enum.reduce(slot_results, {[], false}, fn {_name, result}, {parts, deg} ->
        case result do
          %SlotResult{outcome: :all_failed} ->
            {parts, true}

          %SlotResult{data: data, used_adapter: adapter, outcome: outcome} ->
            slot  = Enum.find(slots, &(&1.name == result.slot))
            entry = Enum.find(slot.chain, &(&1.adapter == adapter))
            body  = apply_field_map(data, entry && entry.field_map)
            is_fallback = outcome == :fallback_used
            {[body | parts], deg || is_fallback}
        end
      end)

    merged_data =
      case strategy do
        :deep_merge    -> Enum.reduce(data_parts, %{}, &deep_merge/2)
        :first_wins    -> List.first(data_parts, %{})
        :field_priority -> field_priority_merge(data_parts)
      end

    %{
      data:    merged_data,
      meta:    build_meta(slot_results, degraded)
    }
  end

  defp deep_merge(left, right) do
    Map.merge(left, right, fn _k, v1, v2 ->
      if is_map(v1) and is_map(v2), do: deep_merge(v1, v2), else: v2
    end)
  end

  # field_priority: later slots can fill in nil fields from earlier slots
  defp field_priority_merge(parts) do
    Enum.reduce(parts, %{}, fn part, acc ->
      Map.merge(acc, part, fn _k, existing, incoming ->
        if is_nil(existing), do: incoming, else: existing
      end)
    end)
  end

  defp apply_field_map(body, nil), do: body || %{}
  defp apply_field_map(body, field_map) do
    Enum.reduce(field_map, body || %{}, fn {old_key, new_key}, acc ->
      case Map.pop(acc, old_key) do
        {nil, acc} -> acc
        {val, acc} -> Map.put(acc, new_key, val)
      end
    end)
  end

  defp build_meta(slot_results, degraded) do
    sources =
      Map.new(slot_results, fn {name, %SlotResult{} = r} ->
        info =
          case r.outcome do
            :ok           -> %{used: r.used_adapter, status: "ok"}
            :fallback_used -> %{used: r.used_adapter, status: "fallback",
                                reason: r.fallback_reason}
            :all_failed   -> %{used: nil, status: "unavailable"}
          end
        {name, info}
      end)

    base = %{sources: sources}
    if degraded, do: Map.put(base, :degraded, true), else: base
  end
end
```

**Response meta example — full scenario (CRM empty, DB fallback used):**

```json
{
  "data": {
    "name": "John Smith",
    "account_no": "ACC-001",
    "balance": 42500
  },
  "meta": {
    "degraded": true,
    "sources": {
      "customer_profile": {
        "used": "local_db",
        "status": "fallback",
        "reason": "empty"
      },
      "account_balance": {
        "used": "banking_legacy",
        "status": "ok"
      }
    }
  }
}
```

**Tests:**
- CRM empty → DB fallback used → `status: "fallback"`, `reason: "empty"` in meta
- All slots ok → no `degraded` key in meta
- One slot `all_failed` (optional) → `degraded: true`, missing slot data absent from `data`
- `field_map` renames key in slot data before merge
- `field_priority` strategy: nil fields filled from fallback, non-nil preserved from primary
- Empty `data_parts` list → `data: {}`

---

### Phase 7.3 — HTTP Response Semantics & Error Envelope

**Goal:** Map slot resolution outcomes to the correct HTTP status codes and consistent
response envelope. Define when fallback usage is transparent vs. surfaced.

#### HTTP Status Mapping

| Outcome | HTTP Status | Reasoning |
|---------|-------------|-----------|
| All slots `:ok` | `200 OK` | Full success |
| Any slot `:fallback_used` (optional slot) | `200 OK` + `meta.degraded: true` | Fallback is transparent to caller; data is complete |
| Any slot `:fallback_used` (required slot) | `200 OK` + `meta.degraded: true` | Same — fallback succeeded, data is returned |
| Any required slot `:all_failed` | `502 Bad Gateway` | No data available from any source in that slot |
| Any optional slot `:all_failed` | `206 Partial Content` + `meta.degraded: true` | Partial data returned |

> **Decision rationale:** When a fallback is used but data was found, the caller gets
> 200 — they asked for data and data was returned. The `meta.sources` section lets
> observability tools track fallback frequency without the caller needing to branch on
> status codes.

#### Error Envelope (502 case)

```json
{
  "error": {
    "code": "UPSTREAM_FAILURE",
    "message": "Required slot(s) could not be resolved: account_balance",
    "meta": {
      "sources": {
        "account_balance": { "used": null, "status": "unavailable" },
        "customer_profile": { "used": "local_db", "status": "fallback", "reason": "timeout" }
      }
    }
  }
}
```

#### Tasks

- `MwKernel.Error.upstream_failure/1` — accepts list of failed slot names
- Gateway controllers: map `{:error, :upstream_failure}` → 502
- Gateway controllers: map partial slot failures → 206 vs 200 per above table
- `MwAudit.Logger`: log `source_meta` (JSON) alongside audit record
- `MwAudit.Logger`: log `fallback_used: true/false` flag on audit record

**Tests:**
- All slots ok → 200, no `degraded` in meta
- Required slot all_failed → 502 with error envelope
- Optional slot all_failed → 206 with `degraded: true`
- Fallback used (required slot) → 200 with `degraded: true`
- Audit record captures `source_meta` and `fallback_used`

---

### Phase 7.4 — Admin UI — Slot-Based Composite Route Editor

**Goal:** Allow ops to create and manage composite routes with slot + chain definitions
through the existing LiveView admin interface.

#### Route Editor LiveView changes (`RouteEditorLive`)

**Route type toggle:** `Single Adapter` → `Composite (Fan-out)`

**Slot management:**
- Add / remove slots (named, required toggle)
- Per-slot chain editor: ordered list of chain entries (drag to reorder priority)
- Per chain entry: adapter selector, timeout_ms, fallback_on checkboxes
  (`:timeout` ✓, `:error` ✓, `:circuit_open` ✓, `:empty` ✓)
- Field map editor: key → renamed key pairs per chain entry

**Merge strategy selector:** `deep_merge` | `field_priority` | `first_wins`

**Live validation:**
- At least one slot defined
- Each slot has at least one chain entry
- At least one slot is required (cannot have all-optional composite)
- No duplicate slot names

**Composite Route List:**
- Badge `FAN-OUT` on route rows
- Expand → shows slot tree with chain entries and priority indicators
- Per-adapter live circuit breaker badge (`:closed` / `:open`) via PubSub

**Deliverable:** Ops can fully configure slot + chain + fallback rules via UI.

**Tests:**
- Creating composite route with 2 slots, each with 1 primary + 1 fallback saves correctly
- Removing all required slots shows validation error
- `fallback_on: :empty` checkbox persists and reloads correctly
- Circuit breaker badge updates live when Fuse trips during test

---

### Phase 7.5 — Observability

**Goal:** Full tracing and metrics for slot resolution, fallback events, and adapter legs.

#### OpenTelemetry Spans

```
[HTTP Request Span]
  ├── [fanout.slot.customer_profile]         duration=3002ms  outcome=fallback_used
  │     ├── [fanout.adapter.crm_api]         duration=3000ms  result=timeout
  │     └── [fanout.adapter.local_db]        duration=1.2ms   result=ok
  └── [fanout.slot.account_balance]          duration=4.3ms   outcome=ok
        └── [fanout.adapter.banking_legacy]  duration=4.3ms   result=ok
```

**Span attributes:**

| Span | Attributes |
|------|-----------|
| `fanout.slot.*` | `slot.name`, `slot.required`, `slot.outcome`, `slot.used_adapter`, `slot.fallback_reason` |
| `fanout.adapter.*` | `adapter.name`, `adapter.priority`, `adapter.result`, `adapter.duration_ms` |

#### Prometheus Metrics

| Metric | Type | Labels |
|--------|------|--------|
| `mw_fanout_requests_total` | counter | `route`, `result` (`ok`/`partial`/`failed`) |
| `mw_fanout_slot_resolution_total` | counter | `slot`, `outcome` (`ok`/`fallback_used`/`all_failed`) |
| `mw_fanout_fallback_total` | counter | `slot`, `primary_adapter`, `fallback_adapter`, `reason` |
| `mw_fanout_adapter_duration_ms` | histogram | `adapter`, `slot`, `result` |

The `mw_fanout_fallback_total` counter is the key operational metric — it shows how often
each primary source is failing and triggering fallback, making source reliability visible.

#### Telemetry Events

```elixir
# Emitted by SlotResolver after resolution
:telemetry.execute([:mw, :fanout, :slot, :resolved], %{duration: dur}, %{
  slot:            slot_name,
  outcome:         outcome,        # :ok | :fallback_used | :all_failed
  used_adapter:    adapter,
  fallback_reason: reason          # nil | :timeout | :error | :circuit_open | :empty
})

# Emitted by FanoutDispatcher per adapter call
:telemetry.execute([:mw, :fanout, :adapter, :stop], %{duration: dur}, %{
  adapter:  name,
  slot:     slot_name,
  result:   :ok | :timeout | :circuit_open | :error
})
```

**Tests:**
- `:mw, :fanout, :slot, :resolved` fires once per slot per request
- `:mw, :fanout, :adapter, :stop` fires once per chain entry per slot
- `outcome: :fallback_used` emitted when CRM empty and DB used
- OTel span `fallback_reason` attribute set correctly on slot span

---

### Phase 7.6 — Integration Tests & Hardening

**Goal:** End-to-end confidence covering all slot/fallback combinations before production.

#### Integration Test Scenarios

| Scenario | Slots | Expected Status | Meta |
|----------|-------|-----------------|------|
| All primary adapters succeed | 2 | 200 | all `status: "ok"` |
| CRM returns `{}` empty, DB fallback has data | 2 | 200 + degraded | `customer_profile: {status: fallback, reason: empty}` |
| CRM times out, DB fallback has data | 2 | 200 + degraded | `reason: timeout` |
| CRM circuit open, DB fallback has data | 2 | 200 + degraded | `reason: circuit_open` |
| CRM fails, DB also fails, required slot | 2 | 502 | upstream_failure |
| CRM fails, DB also fails, optional slot | 2 | 206 + degraded | unavailable in meta |
| Both CRM and DB succeed; priority 1 wins | 1 | 200 | CRM used, not DB |
| All chain entries in all slots succeed | 3 | 200 | all ok |
| Idempotency key replay | 1 | 200 | Cached merged response, no re-dispatch |
| Rate limit exceeded | — | 429 | Before fanout |
| Auth fails | — | 401 | Before fanout |
| Composite route inactive | — | 404 | |
| `field_map` renames `userId` → `user_id` | 1 | 200 | Renamed in merged data |
| `field_priority` strategy: nil field filled from fallback | 1 | 200 + degraded | fallback fills gap |

#### Concurrency & Leak Tests

- 50 concurrent composite requests, each with 2 slots × 2 chain entries (4 tasks/request)
- Verify process count stable (no leaks via `:erlang.system_info(:process_count)`)
- Verify all tasks killed on timeout under load (no zombie tasks)
- Verify `Fuse` circuit state consistent under concurrent access

#### Deliverable

- All integration scenarios pass
- No process leaks under concurrent load
- `mix credo --strict` clean on all new modules
- Dialyzer specs complete: `FanoutDispatcher`, `SlotResolver`, `Merger`
- Test coverage ≥ 80% across new modules

---

## File Map

### New Files

```
apps/mw_kernel/lib/mw_kernel/composite_route.ex     — Slot, ChainEntry, CompositeRoute types
apps/mw_kernel/lib/mw_kernel/slot_result.ex          — SlotResult type
apps/mw_router/lib/mw_router/fanout_dispatcher.ex    — Parallel slot + adapter dispatch
apps/mw_router/lib/mw_router/slot_resolver.ex        — Priority + fallback selection per slot
apps/mw_transform/lib/mw_transform/merger.ex         — Merge slot results into unified payload
```

### Modified Files

```
apps/mw_kernel/lib/mw_kernel/context.ex              — add slot_results, source_meta
apps/mw_router/lib/mw_router/pipeline.ex             — branch on :composite route type
apps/mw_router/lib/mw_router/route_table.ex          — accept composite_spec in route entry
apps/mw_transform/lib/mw_transform/mapper.ex         — expose apply_field_map for reuse
apps/gateway_api/lib/gateway_api_web/controllers/    — 200/206/502 status mapping
apps/gateway_web/lib/gateway_web_web/live/route_editor_live.ex  — slot + chain UI
apps/infra_telemetry/lib/infra_telemetry/metrics.ex  — fanout + fallback metrics
apps/mw_audit/lib/mw_audit/logger.ex                 — source_meta + fallback_used fields
```

### Test Files

```
apps/mw_kernel/test/mw_kernel/composite_route_test.exs
apps/mw_kernel/test/mw_kernel/slot_result_test.exs
apps/mw_router/test/mw_router/fanout_dispatcher_test.exs
apps/mw_router/test/mw_router/slot_resolver_test.exs
apps/mw_transform/test/mw_transform/merger_test.exs
apps/gateway_api/test/integration/composite_route_integration_test.exs
```

---

## Dependencies

No new mix dependencies required:

| Capability | Library | Already in mix.exs |
|---|---|---|
| Parallel tasks | `Task.async_stream` (stdlib) | ✅ |
| Per-adapter timeout | `Task.yield/2` + `Task.shutdown/1` (stdlib) | ✅ |
| Circuit breaker | `fuse ~> 2.5` | ✅ |
| Distributed tracing | `opentelemetry ~> 1.3` | ✅ |
| Metrics | `telemetry_metrics_prometheus_core` | ✅ |

---

## Definition of Done

- [ ] All 6 sub-phases completed and tested
- [ ] Integration test matrix (Phase 7.6) fully passing
- [ ] `mix credo --strict` passes on all new modules
- [ ] Dialyzer specs complete: `FanoutDispatcher`, `SlotResolver`, `Merger`
- [ ] Test coverage ≥ 80% for new modules
- [ ] Admin UI: slot + chain CRUD working in dev
- [ ] OTel waterfall (slot + adapter spans) visible in local Jaeger
- [ ] `mw_fanout_fallback_total` metric firing and queryable in Prometheus
- [ ] Audit records contain `source_meta` and `fallback_used` fields

---

## Open Questions

| # | Question | Owner | Resolution |
|---|----------|-------|------------|
| 1 | Fallback usage: return 200 or 206? | Product | **200 + `meta.degraded: true`** — caller gets complete data, observability layer tracks fallback rate via metrics. 206 reserved for optional-slot all-failed only. |
| 2 | Should `:empty` detection support custom predicates (e.g. `found: false` in body)? | Engineering | Phase 7: detect nil / `%{}` / `[]`. Custom predicate via MFA hook deferred to Phase 8. |
| 3 | Should composite routes support write fan-out (POST to multiple adapters)? | Architecture | Out of scope Phase 7. Separate phase if needed. |
| 4 | Should field_map support nested key paths (`"address.city"`)? | Engineering | Start flat in Phase 7. Dot-notation paths deferred. |
| 5 | Hedged parallel: fire fallback early if primary hasn't responded by N% of its timeout? | Engineering | Not in Phase 7. Current parallel-fire strategy already minimises latency. Add hedging in Phase 8 if metrics show it's needed. |