# Phase 9 — DAG / Conditional Execution Engine

## Overview

Phase 9 activates the **Decision node** (greyed out in Phase 8) and introduces a
full **Directed Acyclic Graph (DAG) execution engine** that replaces the linear
slot-chain model from Phase 7 with a graph-walk interpreter. Flows designed in the
Phase 8 visual canvas — including conditional branches, inter-slot dependencies, and
parallel fan-out — are now executed by the runtime, not just stored.

This phase also adds **historical execution replay**: past request traces can be
replayed on the flow canvas with per-node timing and outcome data, pulled from the
audit log.

---

## What Changes From Phase 7/8

```
Phase 7  → Linear slot chain: Request → [Slot₁ ‖ Slot₂] → Merge → Response
Phase 8  → Visual canvas to design + persist any graph shape
Phase 9  → Runtime that executes the graph exactly as drawn, including
           Decision branches, slot dependencies, and conditional skips
```

Key additions:
- **Decision node execution** — evaluates a predicate against request data or a prior
  slot result; routes execution to the YES branch or NO branch
- **Slot dependency edges** — Slot B can declare it needs Slot A's result before it
  starts; the DAG engine resolves ordering automatically via topological sort
- **DAG executor** — graph-walk loop replacing `FanoutDispatcher`'s static fan-out;
  works for any acyclic topology including diamonds, merges, and conditional skips
- **Conditional slot skip** — if a Decision node routes around a slot, that slot is
  not called and its result is treated as `%SlotResult{status: :skipped}`
- **Historical replay** — load a past request trace from the audit log and step
  through it on the flow canvas (read-only overlay, not a re-execution)

---

## Node Type Activation

| Node Type  | Phase 8 Status       | Phase 9 Status |
|------------|----------------------|----------------|
| Request    | ✅ Full              | unchanged      |
| Slot       | ✅ Full              | + dependency edges |
| Adapter    | ✅ Full              | unchanged      |
| Transform  | ✅ Full              | unchanged      |
| Merge      | ✅ Full              | + conditional merge (skip absent slots) |
| Response   | ✅ Full              | unchanged      |
| Decision   | 🔲 Greyed out (Phase 9) | ✅ Activated  |

---

## Decision Node Specification

A Decision node evaluates a single **predicate** against a **data source** and routes
execution to exactly one of two outgoing paths.

### Property Panel Fields (Phase 8 canvas — activated in Phase 9)

| Field           | Type     | Description |
|-----------------|----------|-------------|
| `data_source`   | select   | `request_body`, `request_header`, `slot_result:<slot_name>` |
| `field_path`    | string   | JSON path into the data source, e.g. `customer.tier` |
| `operator`      | select   | `eq`, `neq`, `gt`, `gte`, `lt`, `lte`, `in`, `not_in`, `present`, `absent` |
| `value`         | string   | Comparison value (string coerced to number for `gt`/`lt`/etc.) |
| `yes_label`     | string   | Label on the YES branch connector (default: "yes") |
| `no_label`      | string   | Label on the NO branch connector (default: "no") |

### Example Decision Config

```elixir
%{
  data_source: "request_body",
  field_path:  "customer.tier",
  operator:    "eq",
  value:       "premium"
}
# YES branch → premium slot chain
# NO  branch → standard slot chain
```

### Connection Rules (extends Phase 8)

```
Decision  ──YES──►  slot, adapter, transform, merge, response
Decision  ──NO───►  slot, adapter, transform, merge, response
Decision cannot have more than one YES or one NO outgoing connection.
Decision cannot be the target of a YES or NO connection (only primary).
```

---

## DAG Execution Engine

### Module: `MwRouter.DagExecutor`

Replaces `FanoutDispatcher` for flows that contain Decision or dependency edges.
`Dispatcher` detects the graph type and routes to the appropriate executor.

```
Has Decision nodes or dependency edges?
  YES → MwRouter.DagExecutor
  NO  → MwRouter.FanoutDispatcher  (unchanged, faster for simple fan-out)
```

### Execution Algorithm

```
1. Topological sort of the node graph (Kahn's algorithm)
2. Build execution layers: nodes in layer N have all dependencies in layer < N
3. Walk layers in order:
   a. For each node in the current layer, check if it should execute:
      - If it is gated by a Decision node, check which branch was taken
      - If gated branch does not include this node, mark as :skipped
   b. Execute non-skipped nodes in the layer concurrently (Task.async_stream)
   c. Collect results into slot_results map
   d. If a Decision node is in this layer, evaluate predicate and record branch
4. After all layers, invoke Merge node with slot_results (skipped slots excluded
   unless required: true, in which case a :slot_missing error is raised)
5. Build Response
```

### Elixir Module Skeleton

```elixir
defmodule MwRouter.DagExecutor do
  @moduledoc """
  Executes a composite flow described by a DAG (nodes + typed edges).

  Handles Decision nodes, slot dependencies, conditional branch skipping,
  and parallel execution within each topological layer.
  """

  alias MwKernel.{CompositeRoute, Context, SlotResult}
  alias MwRouter.SlotResolver

  @type execution_result :: %{
    slot_results:   %{String.t() => SlotResult.t()},
    branch_taken:   %{String.t() => :yes | :no},   # decision_node_id → branch
    skipped_nodes:  MapSet.t(String.t()),
    duration_ms:    non_neg_integer()
  }

  @spec execute(CompositeRoute.t(), Context.t()) ::
          {:ok, execution_result()} | {:error, term()}
  def execute(%CompositeRoute{} = route, %Context{} = ctx) do
    with {:ok, layers} <- topological_sort(route.nodes, route.connections),
         {:ok, result} <- walk_layers(layers, route, ctx) do
      {:ok, result}
    end
  end

  defp topological_sort(nodes, connections) do
    # Kahn's algorithm — returns {:ok, layers} or {:error, :cycle_detected}
  end

  defp walk_layers(layers, route, ctx) do
    # Iterates layers, executing each node respecting skip/branch rules
  end

  defp evaluate_decision(decision_node, ctx, slot_results) do
    # Evaluates predicate → {:yes | :no}
  end
end
```

---

## Slot Dependency Edges

A new connection type `dependency` (dotted grey, already styled in Phase 8) is now
executable: it means "Slot B cannot start until Slot A has a result".

The DAG executor uses dependency edges in the topological sort — Slot B is placed
in a later layer than Slot A automatically. No config is needed beyond drawing the
edge in the canvas.

**Canvas property panel addition (Dependency connections):**

| Field        | Type     | Description |
|--------------|----------|-------------|
| `use_result` | boolean  | If true, Slot B receives Slot A's result as an input param |
| `on_skip`    | select   | `skip_b`, `run_anyway` — what to do if Slot A was skipped |

---

## Conditional Merge

The `Merge` node in Phase 7/8 required all slot results to be present. Phase 9
makes it conditional-aware:

```elixir
defmodule MwTransform.ConditionalMerger do
  @doc """
  Merges slot_results, skipping :skipped entries.
  Raises if a required slot is :skipped (unless skip_required: :allow is set).
  """
  def merge(slot_results, slots_spec, strategy, opts \\ []) do
    # ...
  end
end
```

---

## Canvas Changes (Phase 8 canvas extended)

### Decision Node Property Panel

The Decision node's property panel (greyed out in Phase 8) is now fully active.
Fields: `data_source`, `field_path`, `operator`, `value`.

### Dependency Edge Drawing

Users can draw a `dependency` edge from any Slot to any other Slot. The canvas
validates:
- Only Slot-to-Slot dependency edges are allowed
- No dependency cycles (checked via topological sort preview)

### New Toolbar Button: Validate DAG

Runs the topological sort client-side (JS) and highlights any cycles in red before
the user tries to save.

---

## Historical Replay

### Flow

```
User clicks "Replay" on a past request in Audit Log
  → AuditLogLive passes request_id to FlowBuilderLive via query param
  → FlowBuilderLive loads execution_trace from audit_events table
  → Deserializes per-node outcomes
  → Pushes execution_overlay events for each node (read-only, no re-execution)
  → Canvas shows the execution path of that specific past request
```

### Audit Event Extension

```elixir
# New fields on audit_events rows for composite requests:
%{
  ...,
  composite_trace: %{
    "nodes" => [
      %{"node_id" => "n3", "status" => "fallback", "duration_ms" => 3001,
        "reason" => "timeout", "adapter" => "CrmAdapter"},
      %{"node_id" => "n4", "status" => "ok",       "duration_ms" => 847,
        "adapter" => "DbAdapter"}
    ],
    "branch_taken" => %{"d1" => "yes"},
    "total_ms"     => 3892
  }
}
```

---

## Sub-Phases

### Phase 9.0 — Decision Node Backend

- `MwKernel.Decision` struct and predicate evaluator
- `MwRouter.DagExecutor` topological sort + layer walk
- Decision config added to `FlowBuilderLive` property panel
- Canvas YES/NO branch drawing enabled

**Deliverable:** A flow with one Decision node executes correctly, routing to YES or
NO branch based on request data.

---

### Phase 9.1 — Slot Dependencies + Full DAG Walk

- Dependency edge execution in `DagExecutor`
- `ConditionalMerger` handles skipped slots
- Canvas dependency edge drawing enabled with cycle detection

**Deliverable:** A 3-slot flow where Slot C depends on Slot A's result executes in
the correct order.

---

### Phase 9.2 — Execution Overlay for DAG Flows

- `DagExecutor` emits telemetry events per node (same schema as `FanoutDispatcher`)
- `FlowBuilderLive` receives events and pushes `execution_overlay` per node
- Decision node shows which branch was taken (highlight YES or NO connector)
- Skipped nodes shown with a distinct `:skipped` visual state (grey, dashed border)

**Deliverable:** Running a request against a DAG flow shows the exact path taken on
the canvas in real time.

---

### Phase 9.3 — Historical Replay

- Audit events extended to store `composite_trace` JSON
- `AuditLogLive` gains "Replay on Canvas" button for composite requests
- `FlowBuilderLive` accepts `?replay=<request_id>` query param, loads trace,
  renders read-only overlay

**Deliverable:** Any past composite request can be replayed on the flow canvas from
the audit log.

---

### Phase 9.4 — Force-Directed Auto-Layout + Polish

- Force-directed layout algorithm as a toggle (vs. top-down layered from Phase 8)
- "Validate DAG" toolbar button with client-side cycle detection
- Decision node canvas styling: diamond shape, colour-coded YES/NO anchor points
- `mw_router` and `mw_transform` slots: `mix credo --strict` clean, `@spec` on all
  public functions

**Deliverable:** All Phase 9 features polished, tested, and documented. Phase 9
definition-of-done satisfied.

---

## File Map

### New Files

```
apps/mw_kernel/lib/mw_kernel/decision.ex            — Decision struct + predicate eval
apps/mw_router/lib/mw_router/dag_executor.ex         — DAG walk engine
apps/mw_router/lib/mw_router/dag_validator.ex        — topological sort + cycle check
apps/mw_transform/lib/mw_transform/conditional_merger.ex  — skip-aware merge
apps/gateway_web/assets/js/hooks/dag_validator.js    — client-side cycle detection
apps/gateway_web/lib/gateway_web_web/live/
  replay_live.ex                                      — replay query param handler
apps/infra_repo/priv/repo/migrations/
  20260430000002_add_composite_trace_to_audit_events.exs
```

### Modified Files

```
apps/mw_router/lib/mw_router/dispatcher.ex          — detect DAG, route to DagExecutor
apps/mw_router/lib/mw_router/fanout_dispatcher.ex   — unchanged (used for simple flows)
apps/gateway_web/lib/gateway_web_web/live/
  flow_builder_live.ex                               — Decision panel, dependency edges,
                                                       replay overlay, DAG validate button
  flow_builder_live.html.heex                        — Decision panel UI, replay badge
apps/gateway_web/assets/js/hooks/flow_canvas.js     — dependency edge drawing,
                                                       skipped-node overlay class
apps/infra_telemetry/lib/infra_telemetry/
  fanout_handler.ex                                  — emit per-node events for DAG paths
```

### Test Files

```
apps/mw_router/test/dag_executor_test.exs
apps/mw_router/test/dag_validator_test.exs
apps/mw_transform/test/conditional_merger_test.exs
apps/gateway_web/test/live/flow_builder_dag_test.exs
```

---

## Dependencies

No new Elixir packages required. All building on:
- `Task.async_stream` — parallel layer execution (already used in FanoutDispatcher)
- Phoenix PubSub — telemetry relay to LiveView (already in Phase 8.4)
- Ecto / InfraRepo — audit event extension (already present)

---

## Definition of Done

- [ ] Decision node evaluates all 8 operators correctly (unit tested)
- [ ] DAG executor produces correct topological order for linear, diamond, and
      multi-merge topologies
- [ ] Cycle detection returns `{:error, :cycle_detected}` with the offending node IDs
- [ ] A flow with 2 Decision nodes and 4 conditional branches executes correctly
- [ ] Skipped slots are excluded from merge result; required skipped slot raises error
- [ ] `DagExecutor` telemetry events match `FanoutDispatcher` schema (same overlay code)
- [ ] Historical replay loads correct overlay from audit log for past requests
- [ ] Canvas "Validate DAG" button highlights cycles before save
- [ ] `mix credo --strict` clean on all new modules
- [ ] All new public functions have `@spec` Dialyzer annotations
- [ ] Test coverage ≥ 80% on `DagExecutor`, `DagValidator`, `ConditionalMerger`

---

## Open Questions

| # | Question | Owner | Resolution |
|---|----------|-------|------------|
| 1 | Should Decision predicates support multiple conditions (AND/OR)? | Product | Phase 9: single predicate only. Compound conditions deferred to Phase 10. |
| 2 | Can a Decision node branch back to an earlier node (creating a loop)? | Engineering | No — DAG by definition. Cycle detection blocks this at canvas save time. |
| 3 | How should skipped required slots surface to the API caller? | Product | Return `206 Partial Content` with `X-Skipped-Slots` header listing skipped required slots. |
| 4 | Should `composite_trace` be stored inline on audit_events or in a separate table? | Engineering | Inline JSON column for Phase 9 (simpler). Extract to separate table if query needs arise. |
| 5 | Force-directed layout: library or hand-rolled? | Engineering | D3-force (already available if D3 is added) or a simple spring-mass hand-roll. Decision deferred to Phase 9.4. |